You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by sa...@apache.org on 2018/06/18 04:23:25 UTC
[incubator-nemo] branch master updated: [NEMO-111] Remove
ExecutionProperty.Key (#33)
This is an automated email from the ASF dual-hosted git repository.
sanha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
The following commit(s) were added to refs/heads/master by this push:
new 8b79671 [NEMO-111] Remove ExecutionProperty.Key (#33)
8b79671 is described below
commit 8b7967180621baf15e45decd6ba8e85768ad0cc4
Author: Jangho Seo <ja...@jangho.io>
AuthorDate: Mon Jun 18 13:23:23 2018 +0900
[NEMO-111] Remove ExecutionProperty.Key (#33)
JIRA: [NEMO-111: Remove enum ExecutionProperty.Key](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-111)
**Major changes:**
- Replaced enum ExecutionProperty.Key with Class<? extends ExecutionProperty>
**Minor changes to note:**
- Removed unnecessary casting for accessing ExecutionProperty value.
- Renamed IREdge#getProperty to IREdge#getPropertyValue
- Renamed IRVertex#getProperty to IRVertex#getPropertyValue
- Renamed RuntimeEdge#getProperty to RuntimeEdge#getPropertyValue
resolves [NEMO-111](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-111)
---
.../java/edu/snu/nemo/common/dag/DAGBuilder.java | 24 +++++-----
.../java/edu/snu/nemo/common/ir/edge/IREdge.java | 15 ++++--
.../ir/edge/executionproperty/CoderProperty.java | 6 +--
.../executionproperty/CompressionProperty.java | 12 ++---
.../DataCommunicationPatternProperty.java | 7 +--
.../executionproperty/DataFlowModelProperty.java | 6 +--
.../edge/executionproperty/DataStoreProperty.java | 6 +--
.../DuplicateEdgeGroupProperty.java | 6 +--
.../executionproperty/KeyExtractorProperty.java | 6 +--
.../MetricCollectionProperty.java | 6 +--
.../executionproperty/PartitionerProperty.java | 6 +--
.../UsedDataHandlingProperty.java | 6 +--
.../EdgeExecutionProperty.java} | 26 ++++------
.../ir/executionproperty/ExecutionProperty.java | 56 ++--------------------
.../ir/executionproperty/ExecutionPropertyMap.java | 46 +++++++++---------
.../VertexExecutionProperty.java} | 26 ++++------
.../edu/snu/nemo/common/ir/vertex/IRVertex.java | 15 ++++--
.../edu/snu/nemo/common/ir/vertex/LoopVertex.java | 10 ++--
.../DynamicOptimizationProperty.java | 6 +--
.../ExecutorPlacementProperty.java | 6 +--
.../executionproperty/ParallelismProperty.java | 6 +--
.../ScheduleGroupIndexProperty.java | 6 +--
.../vertex/executionproperty/StageIdProperty.java | 6 +--
.../ExecutionPropertyMapTest.java | 25 +++++-----
.../frontend/spark/core/rdd/PairRDDFunctions.scala | 5 +-
.../compiler/frontend/spark/core/rdd/RDD.scala | 6 +--
.../pass/compiletime/CompileTimePass.java | 2 +-
.../compiletime/annotating/AnnotatingPass.java | 19 ++++----
.../compiletime/annotating/CompressionPass.java | 16 +++----
.../annotating/DataSkewEdgeDataStorePass.java | 3 +-
.../DataSkewEdgeMetricCollectionPass.java | 10 ++--
.../annotating/DataSkewEdgePartitionerPass.java | 5 +-
.../compiletime/annotating/DataSkewVertexPass.java | 3 +-
.../annotating/DefaultEdgeCoderPass.java | 7 +--
.../DefaultEdgeUsedDataHandlingPass.java | 7 ++-
.../annotating/DefaultParallelismPass.java | 23 ++++-----
.../annotating/DefaultStagePartitioningPass.java | 29 ++++++-----
.../DisaggregationEdgeDataStorePass.java | 5 +-
.../annotating/DuplicateEdgeGroupSizePass.java | 23 ++++-----
.../annotating/PadoEdgeDataFlowModelPass.java | 9 ++--
.../annotating/PadoEdgeDataStorePass.java | 18 +++----
.../PadoVertexExecutorPlacementPass.java | 14 ++----
.../ReviseInterStageEdgeDataStorePass.java | 15 +++---
.../annotating/SailfishEdgeDataFlowModelPass.java | 5 +-
.../annotating/SailfishEdgeDataStorePass.java | 7 ++-
.../SailfishEdgeUsedDataHandlingPass.java | 5 +-
.../compiletime/annotating/ScheduleGroupPass.java | 45 +++++++++--------
.../annotating/ShuffleEdgePushPass.java | 5 +-
.../pass/compiletime/composite/CompositePass.java | 4 +-
.../CommonSubexpressionEliminationPass.java | 12 +++--
.../reshaping/DataSkewReshapingPass.java | 11 ++---
.../compiletime/reshaping/LoopExtractionPass.java | 18 +++----
.../compiletime/reshaping/LoopOptimizations.java | 18 +++----
.../pass/compiletime/reshaping/ReshapingPass.java | 6 +--
.../reshaping/SailfishRelayReshapingPass.java | 9 ++--
.../compiler/optimizer/policy/PolicyBuilder.java | 22 +++++----
.../runtime/common/optimizer/RuntimeOptimizer.java | 3 +-
.../runtime/common/plan/PhysicalPlanGenerator.java | 42 ++++++++--------
.../snu/nemo/runtime/common/plan/RuntimeEdge.java | 8 +++-
.../edu/snu/nemo/runtime/executor/Executor.java | 12 +++--
.../runtime/executor/data/SerializerManager.java | 18 +++++--
.../streamchainer/CompressionStreamChainer.java | 4 +-
.../runtime/executor/datatransfer/InputReader.java | 41 ++++++++--------
.../executor/datatransfer/OutputWriter.java | 38 +++++++--------
.../nemo/runtime/executor/data/BlockStoreTest.java | 2 +-
.../executor/datatransfer/DataTransferTest.java | 20 ++++----
.../annotating/DefaultEdgeCoderPassTest.java | 16 +++----
.../annotating/DefaultParallelismPassTest.java | 12 ++---
.../annotating/ScheduleGroupPassTest.java | 9 ++--
.../composite/DataSkewCompositePassTest.java | 15 +++---
.../composite/DisaggregationPassTest.java | 12 ++---
.../composite/PadoCompositePassTest.java | 33 +++++++------
.../compiletime/composite/SailfishPassTest.java | 23 +++++----
.../compiletime/reshaping/LoopFusionPassTest.java | 10 ++--
.../reshaping/LoopInvariantCodeMotionPassTest.java | 8 ++--
75 files changed, 501 insertions(+), 551 deletions(-)
diff --git a/common/src/main/java/edu/snu/nemo/common/dag/DAGBuilder.java b/common/src/main/java/edu/snu/nemo/common/dag/DAGBuilder.java
index 2d90388..ed051f7 100644
--- a/common/src/main/java/edu/snu/nemo/common/dag/DAGBuilder.java
+++ b/common/src/main/java/edu/snu/nemo/common/dag/DAGBuilder.java
@@ -21,9 +21,10 @@ import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
import edu.snu.nemo.common.ir.vertex.SourceVertex;
import edu.snu.nemo.common.ir.vertex.LoopVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.exception.IllegalVertexOperationException;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
import java.io.Serializable;
import java.util.*;
@@ -244,16 +245,16 @@ public final class DAGBuilder<V extends Vertex, E extends Edge<V>> implements Se
// SideInput is not compatible with Push
vertices.forEach(v -> incomingEdges.get(v).stream().filter(e -> e instanceof IREdge).map(e -> (IREdge) e)
.filter(e -> Boolean.TRUE.equals(e.isSideInput()))
- .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getProperty(ExecutionProperty.Key.DataFlowModel)))
+ .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getPropertyValue(DataFlowModelProperty.class).get()))
.forEach(e -> {
throw new RuntimeException("DAG execution property check: "
+ "SideInput edge is not compatible with push" + e.getId());
}));
// DataSizeMetricCollection is not compatible with Push (All data have to be stored before the data collection)
vertices.forEach(v -> incomingEdges.get(v).stream().filter(e -> e instanceof IREdge).map(e -> (IREdge) e)
- .filter(e -> MetricCollectionProperty.Value.DataSkewRuntimePass
- .equals(e.getProperty(ExecutionProperty.Key.MetricCollection)))
- .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getProperty(ExecutionProperty.Key.DataFlowModel)))
+ .filter(e -> Optional.of(MetricCollectionProperty.Value.DataSkewRuntimePass)
+ .equals(e.getPropertyValue(MetricCollectionProperty.class)))
+ .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getPropertyValue(DataFlowModelProperty.class).get()))
.forEach(e -> {
throw new RuntimeException("DAG execution property check: "
+ "DataSizeMetricCollection edge is not compatible with push" + e.getId());
@@ -263,13 +264,14 @@ public final class DAGBuilder<V extends Vertex, E extends Edge<V>> implements Se
vertices.stream().filter(v -> v instanceof IRVertex)
.map(v -> (IRVertex) v)
.forEach(v -> {
- final Integer stageId = v.getProperty(ExecutionProperty.Key.StageId);
- if (stageId != null) {
- if (!stageIdToParallelismMap.containsKey(stageId)) {
- stageIdToParallelismMap.put(stageId, v.getProperty(ExecutionProperty.Key.Parallelism));
- } else if (!stageIdToParallelismMap.get(stageId).equals(v.getProperty(ExecutionProperty.Key.Parallelism))) {
+ final Optional<Integer> stageId = v.getPropertyValue(StageIdProperty.class);
+ if (stageId.isPresent()) {
+ if (!stageIdToParallelismMap.containsKey(stageId.get())) {
+ stageIdToParallelismMap.put(stageId.get(), v.getPropertyValue(ParallelismProperty.class).get());
+ } else if (!stageIdToParallelismMap.get(stageId.get())
+ .equals(v.getPropertyValue(ParallelismProperty.class).get())) {
throw new RuntimeException("DAG execution property check: vertices are in a same stage, "
- + "but has different parallelism execution properties: Stage" + stageId + ": " + v.getId());
+ + "but has different parallelism execution properties: Stage" + stageId.get() + ": " + v.getId());
}
}
});
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/IREdge.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/IREdge.java
index 0778b63..b801841 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/IREdge.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/IREdge.java
@@ -18,17 +18,21 @@ package edu.snu.nemo.common.ir.edge;
import edu.snu.nemo.common.dag.Edge;
import edu.snu.nemo.common.ir.IdManager;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
import edu.snu.nemo.common.ir.executionproperty.ExecutionPropertyMap;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import org.apache.commons.lang3.builder.HashCodeBuilder;
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.function.Consumer;
+
/**
* Physical execution plan of intermediate data movement.
*/
public final class IREdge extends Edge<IRVertex> {
- private final ExecutionPropertyMap executionProperties;
+ private final ExecutionPropertyMap<EdgeExecutionProperty> executionProperties;
private final Boolean isSideInput;
/**
@@ -68,7 +72,7 @@ public final class IREdge extends Edge<IRVertex> {
* @param executionProperty the execution property.
* @return the IREdge with the execution property set.
*/
- public IREdge setProperty(final ExecutionProperty<?> executionProperty) {
+ public IREdge setProperty(final EdgeExecutionProperty<?> executionProperty) {
executionProperties.put(executionProperty);
return this;
}
@@ -80,7 +84,8 @@ public final class IREdge extends Edge<IRVertex> {
* @param executionPropertyKey key of the execution property.
* @return the execution property.
*/
- public <T> T getProperty(final ExecutionProperty.Key executionPropertyKey) {
+ public <T extends Serializable> Optional<T> getPropertyValue(
+ final Class<? extends EdgeExecutionProperty<T>> executionPropertyKey) {
return executionProperties.get(executionPropertyKey);
}
@@ -112,7 +117,7 @@ public final class IREdge extends Edge<IRVertex> {
* @param thatEdge the edge to copy executionProperties to.
*/
public void copyExecutionPropertiesTo(final IREdge thatEdge) {
- this.getExecutionProperties().forEachProperties(thatEdge::setProperty);
+ this.getExecutionProperties().forEachProperties((Consumer<EdgeExecutionProperty>) thatEdge::setProperty);
}
@Override
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CoderProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CoderProperty.java
index 1d2e5dc..aae1c6e 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CoderProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CoderProperty.java
@@ -16,19 +16,19 @@
package edu.snu.nemo.common.ir.edge.executionproperty;
import edu.snu.nemo.common.coder.Coder;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* Coder ExecutionProperty.
*/
-public final class CoderProperty extends ExecutionProperty<Coder> {
+public final class CoderProperty extends EdgeExecutionProperty<Coder> {
/**
* Constructor.
*
* @param value value of the execution property.
*/
private CoderProperty(final Coder value) {
- super(Key.Coder, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CompressionProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CompressionProperty.java
index f5bfbba..6402319 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CompressionProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/CompressionProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* Compression ExecutionProperty.
*/
-public final class CompressionProperty extends ExecutionProperty<CompressionProperty.Compression> {
+public final class CompressionProperty extends EdgeExecutionProperty<CompressionProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
- private CompressionProperty(final Compression value) {
- super(Key.Compression, value);
+ private CompressionProperty(final Value value) {
+ super(value);
}
/**
@@ -34,14 +34,14 @@ public final class CompressionProperty extends ExecutionProperty<CompressionProp
* @param value value of the new execution property.
* @return the newly created execution property.
*/
- public static CompressionProperty of(final Compression value) {
+ public static CompressionProperty of(final Value value) {
return new CompressionProperty(value);
}
/**
* Possible values of Compression ExecutionProperty.
*/
- public enum Compression {
+ public enum Value {
Gzip,
LZ4,
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataCommunicationPatternProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataCommunicationPatternProperty.java
index 39d82c4..cab9b77 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataCommunicationPatternProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataCommunicationPatternProperty.java
@@ -15,19 +15,20 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
+// TODO #492: modularizing runtime components for data communication pattern.
/**
* DataCommunicationPattern ExecutionProperty.
*/
public final class DataCommunicationPatternProperty
- extends ExecutionProperty<DataCommunicationPatternProperty.Value> {
+ extends EdgeExecutionProperty<DataCommunicationPatternProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private DataCommunicationPatternProperty(final Value value) {
- super(Key.DataCommunicationPattern, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataFlowModelProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataFlowModelProperty.java
index 7325384..1c1a8bb 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataFlowModelProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataFlowModelProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* DataFlowModel ExecutionProperty.
*/
-public final class DataFlowModelProperty extends ExecutionProperty<DataFlowModelProperty.Value> {
+public final class DataFlowModelProperty extends EdgeExecutionProperty<DataFlowModelProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private DataFlowModelProperty(final Value value) {
- super(Key.DataFlowModel, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataStoreProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataStoreProperty.java
index d07c934..6e87844 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataStoreProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DataStoreProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* DataStore ExecutionProperty.
*/
-public final class DataStoreProperty extends ExecutionProperty<DataStoreProperty.Value> {
+public final class DataStoreProperty extends EdgeExecutionProperty<DataStoreProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private DataStoreProperty(final Value value) {
- super(Key.DataStore, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DuplicateEdgeGroupProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DuplicateEdgeGroupProperty.java
index 80f0757..8a5f980 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DuplicateEdgeGroupProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/DuplicateEdgeGroupProperty.java
@@ -15,19 +15,19 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* Invariant data ExecutionProperty. Use to indicate same data edge when unrolling loop vertex.
* See DuplicateEdgeGroupPropertyValue
*/
-public final class DuplicateEdgeGroupProperty extends ExecutionProperty<DuplicateEdgeGroupPropertyValue> {
+public final class DuplicateEdgeGroupProperty extends EdgeExecutionProperty<DuplicateEdgeGroupPropertyValue> {
/**
* Constructor.
* @param value value of the execution property.
*/
private DuplicateEdgeGroupProperty(final DuplicateEdgeGroupPropertyValue value) {
- super(Key.DuplicateEdgeGroup, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/KeyExtractorProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/KeyExtractorProperty.java
index 0387e07..7adfc29 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/KeyExtractorProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/KeyExtractorProperty.java
@@ -16,19 +16,19 @@
package edu.snu.nemo.common.ir.edge.executionproperty;
import edu.snu.nemo.common.KeyExtractor;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* KeyExtractor ExecutionProperty.
*/
-public final class KeyExtractorProperty extends ExecutionProperty<KeyExtractor> {
+public final class KeyExtractorProperty extends EdgeExecutionProperty<KeyExtractor> {
/**
* Constructor.
*
* @param value value of the execution property.
*/
private KeyExtractorProperty(final KeyExtractor value) {
- super(Key.KeyExtractor, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/MetricCollectionProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/MetricCollectionProperty.java
index 713be99..d389279 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/MetricCollectionProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/MetricCollectionProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* MetricCollection ExecutionProperty.
*/
-public final class MetricCollectionProperty extends ExecutionProperty<MetricCollectionProperty.Value> {
+public final class MetricCollectionProperty extends EdgeExecutionProperty<MetricCollectionProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private MetricCollectionProperty(final Value value) {
- super(Key.MetricCollection, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/PartitionerProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/PartitionerProperty.java
index 08e607a..249433e 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/PartitionerProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/PartitionerProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* Partitioner ExecutionProperty.
*/
-public final class PartitionerProperty extends ExecutionProperty<PartitionerProperty.Value> {
+public final class PartitionerProperty extends EdgeExecutionProperty<PartitionerProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private PartitionerProperty(final Value value) {
- super(Key.Partitioner, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/UsedDataHandlingProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/UsedDataHandlingProperty.java
index 5b01623..33f6f74 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/UsedDataHandlingProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/edge/executionproperty/UsedDataHandlingProperty.java
@@ -15,19 +15,19 @@
*/
package edu.snu.nemo.common.ir.edge.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
/**
* UsedDataHandling ExecutionProperty.
* This property represents the used data handling strategy.
*/
-public final class UsedDataHandlingProperty extends ExecutionProperty<UsedDataHandlingProperty.Value> {
+public final class UsedDataHandlingProperty extends EdgeExecutionProperty<UsedDataHandlingProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private UsedDataHandlingProperty(final Value value) {
- super(Key.UsedDataHandling, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/EdgeExecutionProperty.java
similarity index 50%
copy from common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
copy to common/src/main/java/edu/snu/nemo/common/ir/executionproperty/EdgeExecutionProperty.java
index e8c03b2..906b72a 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/EdgeExecutionProperty.java
@@ -13,28 +13,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package edu.snu.nemo.common.ir.vertex.executionproperty;
+package edu.snu.nemo.common.ir.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import java.io.Serializable;
/**
- * StageId ExecutionProperty.
+ * {@link ExecutionProperty} for {@link edu.snu.nemo.common.ir.edge.IREdge}.
+ * @param <T> Type of the value.
*/
-public final class StageIdProperty extends ExecutionProperty<Integer> {
+public abstract class EdgeExecutionProperty<T extends Serializable> extends ExecutionProperty<T> {
/**
- * Constructor.
- * @param value value of the execution property.
+ * Default constructor.
+ * @param value value of the EdgeExecutionProperty.
*/
- private StageIdProperty(final Integer value) {
- super(Key.StageId, value);
- }
-
- /**
- * Static method exposing the constructor.
- * @param value value of the new execution property.
- * @return the newly created execution property.
- */
- public static StageIdProperty of(final Integer value) {
- return new StageIdProperty(value);
+ public EdgeExecutionProperty(final T value) {
+ super(value);
}
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionProperty.java
index 45249ae..87efce6 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionProperty.java
@@ -16,23 +16,19 @@
package edu.snu.nemo.common.ir.executionproperty;
import java.io.Serializable;
-import java.util.Objects;
/**
* An abstract class for each execution factors.
- * @param <T> Key of the value.
+ * @param <T> Type of the value.
*/
-public abstract class ExecutionProperty<T> implements Serializable {
- private Key key;
+public abstract class ExecutionProperty<T extends Serializable> implements Serializable {
private T value;
/**
* Default constructor.
- * @param key key of the ExecutionProperty, given by the enum in this class.
* @param value value of the ExecutionProperty.
*/
- public ExecutionProperty(final Key key, final T value) {
- this.key = key;
+ public ExecutionProperty(final T value) {
this.value = value;
}
@@ -43,23 +39,6 @@ public abstract class ExecutionProperty<T> implements Serializable {
return this.value;
}
- /**
- * @return the key of the execution property.
- */
- public final Key getKey() {
- return key;
- }
-
- /**
- * Static method to get an empty execution property.
- * @param <T> type of the value of the execution property.
- * @return an empty execution property.
- */
- static <T> ExecutionProperty<T> emptyExecutionProperty() {
- return new ExecutionProperty<T>(null, null) {
- };
- }
-
@Override
public final boolean equals(final Object o) {
if (this == o) {
@@ -69,36 +48,11 @@ public abstract class ExecutionProperty<T> implements Serializable {
return false;
}
final ExecutionProperty<?> that = (ExecutionProperty<?>) o;
- return getKey() == that.getKey()
- && Objects.equals(getValue(), that.getValue());
+ return value != null ? value.equals(that.value) : that.value == null;
}
@Override
public final int hashCode() {
- return Objects.hash(getKey(), getValue());
- }
-
- /**
- * Key for different types of execution property.
- */
- public enum Key {
- // Applies to IREdge
- DataCommunicationPattern,
- DataFlowModel,
- DataStore,
- MetricCollection,
- Partitioner,
- KeyExtractor,
- UsedDataHandling,
- Compression,
- DuplicateEdgeGroup,
- Coder,
-
- // Applies to IRVertex
- DynamicOptimizationType,
- ExecutorPlacement,
- Parallelism,
- ScheduleGroupIndex,
- StageId,
+ return value != null ? value.hashCode() : 0;
}
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMap.java b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMap.java
index c3b2e22..c4b297b 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMap.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMap.java
@@ -28,6 +28,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
+import javax.annotation.concurrent.NotThreadSafe;
import java.io.Serializable;
import java.util.*;
import java.util.function.Consumer;
@@ -35,10 +36,12 @@ import java.util.stream.Collectors;
/**
* ExecutionPropertyMap Class, which uses HashMap for keeping track of ExecutionProperties for vertices and edges.
+ * @param <T> Type of {@link ExecutionProperty} this map stores.
*/
-public final class ExecutionPropertyMap implements Serializable {
+@NotThreadSafe
+public final class ExecutionPropertyMap<T extends ExecutionProperty> implements Serializable {
private final String id;
- private final Map<ExecutionProperty.Key, ExecutionProperty<?>> properties;
+ private final Map<Class<? extends ExecutionProperty>, T> properties = new HashMap<>();
/**
* Constructor for ExecutionPropertyMap class.
@@ -47,7 +50,6 @@ public final class ExecutionPropertyMap implements Serializable {
@VisibleForTesting
public ExecutionPropertyMap(final String id) {
this.id = id;
- properties = new EnumMap<>(ExecutionProperty.Key.class);
}
/**
@@ -56,9 +58,10 @@ public final class ExecutionPropertyMap implements Serializable {
* @param commPattern Data communication pattern type of the edge.
* @return The corresponding ExecutionPropertyMap.
*/
- public static ExecutionPropertyMap of(final IREdge irEdge,
- final DataCommunicationPatternProperty.Value commPattern) {
- final ExecutionPropertyMap map = new ExecutionPropertyMap(irEdge.getId());
+ public static ExecutionPropertyMap<EdgeExecutionProperty> of(
+ final IREdge irEdge,
+ final DataCommunicationPatternProperty.Value commPattern) {
+ final ExecutionPropertyMap<EdgeExecutionProperty> map = new ExecutionPropertyMap<>(irEdge.getId());
map.put(DataCommunicationPatternProperty.of(commPattern));
map.put(DataFlowModelProperty.of(DataFlowModelProperty.Value.Pull));
switch (commPattern) {
@@ -85,8 +88,8 @@ public final class ExecutionPropertyMap implements Serializable {
* @param irVertex irVertex to keep the execution property of.
* @return The corresponding ExecutionPropertyMap.
*/
- public static ExecutionPropertyMap of(final IRVertex irVertex) {
- final ExecutionPropertyMap map = new ExecutionPropertyMap(irVertex.getId());
+ public static ExecutionPropertyMap<VertexExecutionProperty> of(final IRVertex irVertex) {
+ final ExecutionPropertyMap<VertexExecutionProperty> map = new ExecutionPropertyMap<>(irVertex.getId());
map.put(ParallelismProperty.of(1));
map.put(ExecutorPlacementProperty.of(ExecutorPlacementProperty.NONE));
return map;
@@ -103,22 +106,21 @@ public final class ExecutionPropertyMap implements Serializable {
/**
* Put the given execution property in the ExecutionPropertyMap.
* @param executionProperty execution property to insert.
- * @return the inserted execution property.
+ * @return the previous execution property, or null if there was no execution property with the specified property key
*/
- public ExecutionProperty<?> put(final ExecutionProperty<?> executionProperty) {
- return properties.put(executionProperty.getKey(), executionProperty);
+ public T put(final T executionProperty) {
+ return properties.put(executionProperty.getClass(), executionProperty);
}
/**
* Get the value of the given execution property type.
- * @param <T> Type of the return value.
+ * @param <U> Type of the return value.
* @param executionPropertyKey the execution property type to find the value of.
* @return the value of the given execution property.
*/
- public <T> T get(final ExecutionProperty.Key executionPropertyKey) {
- ExecutionProperty<T> property = (ExecutionProperty<T>) properties.getOrDefault(executionPropertyKey,
- ExecutionProperty.<T>emptyExecutionProperty());
- return property.getValue();
+ public <U extends Serializable> Optional<U> get(final Class<? extends ExecutionProperty<U>> executionPropertyKey) {
+ final ExecutionProperty<U> property = properties.get(executionPropertyKey);
+ return property == null ? Optional.empty() : Optional.of(property.getValue());
}
/**
@@ -126,7 +128,7 @@ public final class ExecutionPropertyMap implements Serializable {
* @param key key of the execution property to remove.
* @return the removed execution property
*/
- public ExecutionProperty<?> remove(final ExecutionProperty.Key key) {
+ public T remove(final Class<? extends T> key) {
return properties.remove(key);
}
@@ -134,7 +136,7 @@ public final class ExecutionPropertyMap implements Serializable {
* @param key key to look for.
* @return whether or not the execution property map contains the key.
*/
- public boolean containsKey(final ExecutionProperty.Key key) {
+ public boolean containsKey(final Class<? extends T> key) {
return properties.containsKey(key);
}
@@ -142,7 +144,7 @@ public final class ExecutionPropertyMap implements Serializable {
* Same as forEach function in Java 8, but for execution properties.
* @param action action to apply to each of the execution properties.
*/
- public void forEachProperties(final Consumer<? super ExecutionProperty> action) {
+ public void forEachProperties(final Consumer<? super T> action) {
properties.values().forEach(action);
}
@@ -151,7 +153,7 @@ public final class ExecutionPropertyMap implements Serializable {
final StringBuilder sb = new StringBuilder();
sb.append("{");
boolean isFirstPair = true;
- for (final Map.Entry<ExecutionProperty.Key, ExecutionProperty<?>> entry : properties.entrySet()) {
+ for (final Map.Entry<Class<? extends ExecutionProperty>, T> entry : properties.entrySet()) {
if (!isFirstPair) {
sb.append(", ");
}
@@ -180,8 +182,8 @@ public final class ExecutionPropertyMap implements Serializable {
ExecutionPropertyMap that = (ExecutionPropertyMap) obj;
return new EqualsBuilder()
- .append(properties.values().stream().map(ExecutionProperty::getValue).collect(Collectors.toSet()),
- that.properties.values().stream().map(ExecutionProperty::getValue).collect(Collectors.toSet()))
+ .append(properties.values().stream().collect(Collectors.toSet()),
+ that.properties.values().stream().collect(Collectors.toSet()))
.isEquals();
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/VertexExecutionProperty.java
similarity index 50%
copy from common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
copy to common/src/main/java/edu/snu/nemo/common/ir/executionproperty/VertexExecutionProperty.java
index e8c03b2..4bdd2e6 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/executionproperty/VertexExecutionProperty.java
@@ -13,28 +13,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package edu.snu.nemo.common.ir.vertex.executionproperty;
+package edu.snu.nemo.common.ir.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import java.io.Serializable;
/**
- * StageId ExecutionProperty.
+ * {@link ExecutionProperty} for {@link edu.snu.nemo.common.ir.vertex.IRVertex}.
+ * @param <T> Type of the value.
*/
-public final class StageIdProperty extends ExecutionProperty<Integer> {
+public abstract class VertexExecutionProperty<T extends Serializable> extends ExecutionProperty<T> {
/**
- * Constructor.
- * @param value value of the execution property.
+ * Default constructor.
+ * @param value value of the VertexExecutionProperty.
*/
- private StageIdProperty(final Integer value) {
- super(Key.StageId, value);
- }
-
- /**
- * Static method exposing the constructor.
- * @param value value of the new execution property.
- * @return the newly created execution property.
- */
- public static StageIdProperty of(final Integer value) {
- return new StageIdProperty(value);
+ public VertexExecutionProperty(final T value) {
+ super(value);
}
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/IRVertex.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/IRVertex.java
index bcaaa9d..de7359b 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/IRVertex.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/IRVertex.java
@@ -18,14 +18,18 @@ package edu.snu.nemo.common.ir.vertex;
import edu.snu.nemo.common.ir.IdManager;
import edu.snu.nemo.common.ir.executionproperty.ExecutionPropertyMap;
import edu.snu.nemo.common.dag.Vertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
+
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.function.Consumer;
/**
* The basic unit of operation in a dataflow program, as well as the most important data structure in Nemo.
* An IRVertex is created and modified in the compiler, and executed in the runtime.
*/
public abstract class IRVertex extends Vertex {
- private final ExecutionPropertyMap executionProperties;
+ private final ExecutionPropertyMap<VertexExecutionProperty> executionProperties;
/**
* Constructor of IRVertex.
@@ -45,7 +49,7 @@ public abstract class IRVertex extends Vertex {
* @param thatVertex the edge to copy executionProperties to.
*/
public final void copyExecutionPropertiesTo(final IRVertex thatVertex) {
- this.getExecutionProperties().forEachProperties(thatVertex::setProperty);
+ this.getExecutionProperties().forEachProperties((Consumer<VertexExecutionProperty>) thatVertex::setProperty);
}
/**
@@ -53,7 +57,7 @@ public abstract class IRVertex extends Vertex {
* @param executionProperty new execution property.
* @return the IRVertex with the execution property set.
*/
- public final IRVertex setProperty(final ExecutionProperty<?> executionProperty) {
+ public final IRVertex setProperty(final VertexExecutionProperty<?> executionProperty) {
executionProperties.put(executionProperty);
return this;
}
@@ -64,7 +68,8 @@ public abstract class IRVertex extends Vertex {
* @param executionPropertyKey key of the execution property.
* @return the execution property.
*/
- public final <T> T getProperty(final ExecutionProperty.Key executionPropertyKey) {
+ public final <T extends Serializable> Optional<T> getPropertyValue(
+ final Class<? extends VertexExecutionProperty<T>> executionPropertyKey) {
return executionProperties.get(executionPropertyKey);
}
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/LoopVertex.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/LoopVertex.java
index 20d2eca..a761c19 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/LoopVertex.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/LoopVertex.java
@@ -18,9 +18,9 @@ package edu.snu.nemo.common.ir.vertex;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupPropertyValue;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import java.io.Serializable;
import java.util.*;
@@ -217,7 +217,7 @@ public final class LoopVertex extends IRVertex {
dagBuilder.addVertex(newIrVertex, dagToAdd);
dagToAdd.getIncomingEdgesOf(irVertex).forEach(edge -> {
final IRVertex newSrc = originalToNewIRVertex.get(edge.getSrc());
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
newSrc, newIrVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
dagBuilder.connectVertices(newIrEdge);
@@ -226,7 +226,7 @@ public final class LoopVertex extends IRVertex {
// process DAG incoming edges.
getDagIncomingEdges().forEach((dstVertex, irEdges) -> irEdges.forEach(edge -> {
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
edge.getSrc(), originalToNewIRVertex.get(dstVertex), edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
dagBuilder.connectVertices(newIrEdge);
@@ -235,7 +235,7 @@ public final class LoopVertex extends IRVertex {
if (loopTerminationConditionMet()) {
// if termination condition met, we process the DAG outgoing edge.
getDagOutgoingEdges().forEach((srcVertex, irEdges) -> irEdges.forEach(edge -> {
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
originalToNewIRVertex.get(srcVertex), edge.getDst(), edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
dagBuilder.addVertex(edge.getDst()).connectVertices(newIrEdge);
@@ -246,7 +246,7 @@ public final class LoopVertex extends IRVertex {
this.getDagIncomingEdges().clear();
this.nonIterativeIncomingEdges.forEach((dstVertex, irEdges) -> irEdges.forEach(this::addDagIncomingEdge));
this.iterativeIncomingEdges.forEach((dstVertex, irEdges) -> irEdges.forEach(edge -> {
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
originalToNewIRVertex.get(edge.getSrc()), dstVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
this.addDagIncomingEdge(newIrEdge);
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/DynamicOptimizationProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/DynamicOptimizationProperty.java
index d0fa611..7fb808f 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/DynamicOptimizationProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/DynamicOptimizationProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.vertex.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
/**
* DynamicOptimizationType ExecutionProperty.
*/
-public final class DynamicOptimizationProperty extends ExecutionProperty<DynamicOptimizationProperty.Value> {
+public final class DynamicOptimizationProperty extends VertexExecutionProperty<DynamicOptimizationProperty.Value> {
/**
* Constructor.
* @param value value of the execution property.
*/
private DynamicOptimizationProperty(final Value value) {
- super(Key.DynamicOptimizationType, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ExecutorPlacementProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ExecutorPlacementProperty.java
index dc2220e..af22ed0 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ExecutorPlacementProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ExecutorPlacementProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.vertex.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
/**
* ExecutionPlacement ExecutionProperty.
*/
-public final class ExecutorPlacementProperty extends ExecutionProperty<String> {
+public final class ExecutorPlacementProperty extends VertexExecutionProperty<String> {
/**
* Constructor.
* @param value value of the execution property.
*/
private ExecutorPlacementProperty(final String value) {
- super(Key.ExecutorPlacement, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ParallelismProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ParallelismProperty.java
index 40d8786..d07731d 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ParallelismProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ParallelismProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.vertex.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
/**
* Parallelism ExecutionProperty.
*/
-public final class ParallelismProperty extends ExecutionProperty<Integer> {
+public final class ParallelismProperty extends VertexExecutionProperty<Integer> {
/**
* Constructor.
* @param value value of the execution property.
*/
private ParallelismProperty(final Integer value) {
- super(Key.Parallelism, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ScheduleGroupIndexProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ScheduleGroupIndexProperty.java
index 2f19b61..08518ff 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ScheduleGroupIndexProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/ScheduleGroupIndexProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.vertex.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
/**
* ScheduleGroupIndex ExecutionProperty.
*/
-public final class ScheduleGroupIndexProperty extends ExecutionProperty<Integer> {
+public final class ScheduleGroupIndexProperty extends VertexExecutionProperty<Integer> {
/**
* Constructor.
* @param value value of the execution property.
*/
private ScheduleGroupIndexProperty(final Integer value) {
- super(Key.ScheduleGroupIndex, value);
+ super(value);
}
/**
diff --git a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
index e8c03b2..0900992 100644
--- a/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
+++ b/common/src/main/java/edu/snu/nemo/common/ir/vertex/executionproperty/StageIdProperty.java
@@ -15,18 +15,18 @@
*/
package edu.snu.nemo.common.ir.vertex.executionproperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
/**
* StageId ExecutionProperty.
*/
-public final class StageIdProperty extends ExecutionProperty<Integer> {
+public final class StageIdProperty extends VertexExecutionProperty<Integer> {
/**
* Constructor.
* @param value value of the execution property.
*/
private StageIdProperty(final Integer value) {
- super(Key.StageId, value);
+ super(value);
}
/**
diff --git a/common/src/test/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMapTest.java b/common/src/test/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMapTest.java
index 6ca8295..f7eaad7 100644
--- a/common/src/test/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMapTest.java
+++ b/common/src/test/java/edu/snu/nemo/common/ir/executionproperty/ExecutionPropertyMapTest.java
@@ -21,8 +21,9 @@ import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
import edu.snu.nemo.common.ir.executionproperty.ExecutionPropertyMap;
+import edu.snu.nemo.common.ir.executionproperty.VertexExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
@@ -31,7 +32,7 @@ import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertFalse;
/**
* Test {@link ExecutionPropertyMap}.
@@ -42,8 +43,8 @@ public class ExecutionPropertyMapTest {
private final DataCommunicationPatternProperty.Value comPattern = DataCommunicationPatternProperty.Value.OneToOne;
private final IREdge edge = new IREdge(DataCommunicationPatternProperty.Value.OneToOne, source, destination);
- private ExecutionPropertyMap edgeMap;
- private ExecutionPropertyMap vertexMap;
+ private ExecutionPropertyMap<EdgeExecutionProperty> edgeMap;
+ private ExecutionPropertyMap<VertexExecutionProperty> vertexMap;
@Before
public void setUp() {
@@ -53,8 +54,8 @@ public class ExecutionPropertyMapTest {
@Test
public void testDefaultValues() {
- assertEquals(comPattern, edgeMap.get(ExecutionProperty.Key.DataCommunicationPattern));
- assertEquals(1, vertexMap.<Integer>get(ExecutionProperty.Key.Parallelism).longValue());
+ assertEquals(comPattern, edgeMap.get(DataCommunicationPatternProperty.class).get());
+ assertEquals(1, vertexMap.get(ParallelismProperty.class).get().longValue());
assertEquals(edge.getId(), edgeMap.getId());
assertEquals(source.getId(), vertexMap.getId());
}
@@ -62,16 +63,16 @@ public class ExecutionPropertyMapTest {
@Test
public void testPutGetAndRemove() {
edgeMap.put(DataStoreProperty.of(DataStoreProperty.Value.MemoryStore));
- assertEquals(DataStoreProperty.Value.MemoryStore, edgeMap.get(ExecutionProperty.Key.DataStore));
+ assertEquals(DataStoreProperty.Value.MemoryStore, edgeMap.get(DataStoreProperty.class).get());
edgeMap.put(DataFlowModelProperty.of(DataFlowModelProperty.Value.Pull));
- assertEquals(DataFlowModelProperty.Value.Pull, edgeMap.get(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataFlowModelProperty.Value.Pull, edgeMap.get(DataFlowModelProperty.class).get());
edgeMap.put(CoderProperty.of(Coder.DUMMY_CODER));
- assertEquals(Coder.DUMMY_CODER, edgeMap.get(ExecutionProperty.Key.Coder));
+ assertEquals(Coder.DUMMY_CODER, edgeMap.get(CoderProperty.class).get());
- edgeMap.remove(ExecutionProperty.Key.DataFlowModel);
- assertNull(edgeMap.get(ExecutionProperty.Key.DataFlowModel));
+ edgeMap.remove(DataFlowModelProperty.class);
+ assertFalse(edgeMap.get(DataFlowModelProperty.class).isPresent());
vertexMap.put(ParallelismProperty.of(100));
- assertEquals(100, vertexMap.<Integer>get(ExecutionProperty.Key.Parallelism).longValue());
+ assertEquals(100, vertexMap.get(ParallelismProperty.class).get().longValue());
}
}
diff --git a/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/PairRDDFunctions.scala b/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/PairRDDFunctions.scala
index 0b61d7b..5065f9c 100644
--- a/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/PairRDDFunctions.scala
+++ b/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/PairRDDFunctions.scala
@@ -20,7 +20,7 @@ import java.util
import edu.snu.nemo.common.dag.DAGBuilder
import edu.snu.nemo.common.ir.edge.IREdge
import edu.snu.nemo.common.ir.edge.executionproperty.{CoderProperty, KeyExtractorProperty}
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty
import edu.snu.nemo.common.ir.vertex.{IRVertex, LoopVertex, OperatorVertex}
import edu.snu.nemo.compiler.frontend.spark.SparkKeyExtractor
import edu.snu.nemo.compiler.frontend.spark.coder.SparkCoder
@@ -72,7 +72,8 @@ final class PairRDDFunctions[K: ClassTag, V: ClassTag] protected[rdd] (
val newEdge = new IREdge(SparkFrontendUtils.getEdgeCommunicationPattern(self.lastVertex, reduceByKeyVertex),
self.lastVertex, reduceByKeyVertex)
newEdge.setProperty(
- CoderProperty.of(new SparkCoder[Tuple2[K, V]](self.serializer)).asInstanceOf[ExecutionProperty[_]])
+ CoderProperty.of(new SparkCoder[Tuple2[K, V]](self.serializer))
+ .asInstanceOf[EdgeExecutionProperty[_ <: Serializable]])
newEdge.setProperty(KeyExtractorProperty.of(new SparkKeyExtractor))
builder.connectVertices(newEdge)
diff --git a/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/RDD.scala b/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/RDD.scala
index 12318c5..135f698 100644
--- a/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/RDD.scala
+++ b/compiler/frontend/spark/src/main/scala/edu/snu/nemo/compiler/frontend/spark/core/rdd/RDD.scala
@@ -21,7 +21,7 @@ import edu.snu.nemo.client.JobLauncher
import edu.snu.nemo.common.dag.{DAG, DAGBuilder}
import edu.snu.nemo.common.ir.edge.IREdge
import edu.snu.nemo.common.ir.edge.executionproperty.{CoderProperty, KeyExtractorProperty}
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty
import edu.snu.nemo.common.ir.vertex.{IRVertex, LoopVertex, OperatorVertex}
import edu.snu.nemo.compiler.frontend.spark.SparkKeyExtractor
import edu.snu.nemo.compiler.frontend.spark.coder.SparkCoder
@@ -51,8 +51,8 @@ final class RDD[T: ClassTag] protected[rdd] (
protected[rdd] val serializer: Serializer = SparkFrontendUtils.deriveSerializerFrom(_sc)
private val loopVertexStack = new util.Stack[LoopVertex]
- private val coderProperty: ExecutionProperty[_] =
- CoderProperty.of(new SparkCoder[T](serializer)).asInstanceOf[ExecutionProperty[_]]
+ private val coderProperty: EdgeExecutionProperty[_ <: Serializable] =
+ CoderProperty.of(new SparkCoder[T](serializer)).asInstanceOf[EdgeExecutionProperty[_ <: Serializable]]
private val keyExtractorProperty: KeyExtractorProperty = KeyExtractorProperty.of(new SparkKeyExtractor)
/**
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/CompileTimePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/CompileTimePass.java
index eab7a29..72cbd01 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/CompileTimePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/CompileTimePass.java
@@ -33,5 +33,5 @@ public interface CompileTimePass extends Function<DAG<IRVertex, IREdge>, DAG<IRV
* Getter for prerequisite execution properties.
* @return set of prerequisite execution properties.
*/
- Set<ExecutionProperty.Key> getPrerequisiteExecutionProperties();
+ Set<Class<? extends ExecutionProperty>> getPrerequisiteExecutionProperties();
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/AnnotatingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/AnnotatingPass.java
index 21aa664..b7d3690 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/AnnotatingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/AnnotatingPass.java
@@ -18,7 +18,7 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.CompileTimePass;
-import java.util.HashSet;
+import java.util.Collections;
import java.util.Set;
/**
@@ -26,16 +26,16 @@ import java.util.Set;
* It is ensured by the compiler that the shape of the IR DAG itself is not modified by an AnnotatingPass.
*/
public abstract class AnnotatingPass implements CompileTimePass {
- private final ExecutionProperty.Key keyOfExecutionPropertyToModify;
- private final Set<ExecutionProperty.Key> prerequisiteExecutionProperties;
+ private final Class<? extends ExecutionProperty> keyOfExecutionPropertyToModify;
+ private final Set<Class<? extends ExecutionProperty>> prerequisiteExecutionProperties;
/**
* Constructor.
* @param keyOfExecutionPropertyToModify key of execution property to modify.
* @param prerequisiteExecutionProperties prerequisite execution properties.
*/
- public AnnotatingPass(final ExecutionProperty.Key keyOfExecutionPropertyToModify,
- final Set<ExecutionProperty.Key> prerequisiteExecutionProperties) {
+ public AnnotatingPass(final Class<? extends ExecutionProperty> keyOfExecutionPropertyToModify,
+ final Set<Class<? extends ExecutionProperty>> prerequisiteExecutionProperties) {
this.keyOfExecutionPropertyToModify = keyOfExecutionPropertyToModify;
this.prerequisiteExecutionProperties = prerequisiteExecutionProperties;
}
@@ -44,21 +44,20 @@ public abstract class AnnotatingPass implements CompileTimePass {
* Constructor.
* @param keyOfExecutionPropertyToModify key of execution property to modify.
*/
- public AnnotatingPass(final ExecutionProperty.Key keyOfExecutionPropertyToModify) {
- this.keyOfExecutionPropertyToModify = keyOfExecutionPropertyToModify;
- this.prerequisiteExecutionProperties = new HashSet<>();
+ public AnnotatingPass(final Class<? extends ExecutionProperty> keyOfExecutionPropertyToModify) {
+ this(keyOfExecutionPropertyToModify, Collections.emptySet());
}
/**
* Getter for key of execution property to modify.
* @return key of execution property to modify.
*/
- public final ExecutionProperty.Key getExecutionPropertyToModify() {
+ public final Class<? extends ExecutionProperty> getExecutionPropertyToModify() {
return keyOfExecutionPropertyToModify;
}
@Override
- public final Set<ExecutionProperty.Key> getPrerequisiteExecutionProperties() {
+ public final Set<Class<? extends ExecutionProperty>> getPrerequisiteExecutionProperties() {
return prerequisiteExecutionProperties;
}
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/CompressionPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/CompressionPass.java
index 5773159..1d1a651 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/CompressionPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/CompressionPass.java
@@ -18,38 +18,38 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.CompressionProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
/**
* A pass for applying compression algorithm for data flowing between vertices.
*/
public final class CompressionPass extends AnnotatingPass {
- private final CompressionProperty.Compression compression;
+ private final CompressionProperty.Value compression;
/**
* Default constructor. Uses LZ4 as default.
*/
public CompressionPass() {
- super(ExecutionProperty.Key.Compression);
- this.compression = CompressionProperty.Compression.LZ4;
+ super(CompressionProperty.class);
+ this.compression = CompressionProperty.Value.LZ4;
}
/**
* Constructor.
* @param compression Compression to apply on edges.
*/
- public CompressionPass(final CompressionProperty.Compression compression) {
- super(ExecutionProperty.Key.Compression);
+ public CompressionPass(final CompressionProperty.Value compression) {
+ super(CompressionProperty.class);
this.compression = compression;
}
@Override
public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
dag.topologicalDo(vertex -> dag.getIncomingEdgesOf(vertex).stream()
- .filter(e -> !vertex.getProperty(ExecutionProperty.Key.StageId)
- .equals(e.getSrc().getProperty(ExecutionProperty.Key.StageId)))
+ .filter(e -> !vertex.getPropertyValue(StageIdProperty.class).get()
+ .equals(e.getSrc().getPropertyValue(StageIdProperty.class).get()))
.forEach(edge -> edge.setProperty(CompressionProperty.of(compression))));
return dag;
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeDataStorePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeDataStorePass.java
index 2c34425..a0428a6 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeDataStorePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeDataStorePass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
/**
@@ -32,7 +31,7 @@ public final class DataSkewEdgeDataStorePass extends AnnotatingPass {
* Default constructor.
*/
public DataSkewEdgeDataStorePass() {
- super(ExecutionProperty.Key.DataStore);
+ super(DataStoreProperty.class);
}
@Override
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeMetricCollectionPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeMetricCollectionPass.java
index 1d04e04..614d81b 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeMetricCollectionPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgeMetricCollectionPass.java
@@ -20,11 +20,9 @@ import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.MetricCollectionProperty;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
+import java.util.Collections;
/**
* Pass to annotate the DAG for a job to perform data skew.
@@ -36,9 +34,7 @@ public final class DataSkewEdgeMetricCollectionPass extends AnnotatingPass {
* Default constructor.
*/
public DataSkewEdgeMetricCollectionPass() {
- super(ExecutionProperty.Key.MetricCollection, Stream.of(
- ExecutionProperty.Key.DataCommunicationPattern
- ).collect(Collectors.toSet()));
+ super(MetricCollectionProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -48,7 +44,7 @@ public final class DataSkewEdgeMetricCollectionPass extends AnnotatingPass {
if (v instanceof MetricCollectionBarrierVertex) {
dag.getOutgoingEdgesOf(v).forEach(edge -> {
// double checking.
- if (edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)
+ if (edge.getPropertyValue(DataCommunicationPatternProperty.class).get()
.equals(DataCommunicationPatternProperty.Value.Shuffle)) {
edge.setProperty(MetricCollectionProperty.of(MetricCollectionProperty.Value.DataSkewRuntimePass));
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgePartitionerPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgePartitionerPass.java
index 71e1161..4fcfe41 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgePartitionerPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewEdgePartitionerPass.java
@@ -20,7 +20,6 @@ import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.MetricCollectionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.PartitionerProperty;
import java.util.Collections;
@@ -34,7 +33,7 @@ public final class DataSkewEdgePartitionerPass extends AnnotatingPass {
* Default constructor.
*/
public DataSkewEdgePartitionerPass() {
- super(ExecutionProperty.Key.Partitioner, Collections.singleton(ExecutionProperty.Key.MetricCollection));
+ super(PartitionerProperty.class, Collections.singleton(MetricCollectionProperty.class));
}
@Override
@@ -45,7 +44,7 @@ public final class DataSkewEdgePartitionerPass extends AnnotatingPass {
outEdges.forEach(edge -> {
// double checking.
if (MetricCollectionProperty.Value.DataSkewRuntimePass
- .equals(edge.getProperty(ExecutionProperty.Key.MetricCollection))) {
+ .equals(edge.getPropertyValue(MetricCollectionProperty.class).get())) {
edge.setProperty(PartitionerProperty.of(PartitionerProperty.Value.DataSkewHashPartitioner));
}
});
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewVertexPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewVertexPass.java
index 6955bca..c9e7aa0 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewVertexPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DataSkewVertexPass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.DynamicOptimizationProperty;
/**
@@ -31,7 +30,7 @@ public final class DataSkewVertexPass extends AnnotatingPass {
* Default constructor.
*/
public DataSkewVertexPass() {
- super(ExecutionProperty.Key.DynamicOptimizationType);
+ super(DynamicOptimizationProperty.class);
}
@Override
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPass.java
index 02ace96..eade93d 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPass.java
@@ -19,11 +19,8 @@ import edu.snu.nemo.common.coder.Coder;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import java.util.Collections;
-
/**
* Pass for initiating IREdge Coder ExecutionProperty with default dummy coder.
*/
@@ -35,14 +32,14 @@ public final class DefaultEdgeCoderPass extends AnnotatingPass {
* Default constructor.
*/
public DefaultEdgeCoderPass() {
- super(ExecutionProperty.Key.Coder, Collections.emptySet());
+ super(CoderProperty.class);
}
@Override
public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
dag.topologicalDo(irVertex ->
dag.getIncomingEdgesOf(irVertex).forEach(irEdge -> {
- if (irEdge.getProperty(ExecutionProperty.Key.Coder) == null) {
+ if (!irEdge.getPropertyValue(CoderProperty.class).isPresent()) {
irEdge.setProperty(DEFAULT_CODER_PROPERTY);
}
}));
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeUsedDataHandlingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeUsedDataHandlingPass.java
index 3acf173..81e16f7 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeUsedDataHandlingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeUsedDataHandlingPass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.UsedDataHandlingProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import java.util.Collections;
@@ -33,15 +32,15 @@ public final class DefaultEdgeUsedDataHandlingPass extends AnnotatingPass {
* Default constructor.
*/
public DefaultEdgeUsedDataHandlingPass() {
- super(ExecutionProperty.Key.UsedDataHandling, Collections.singleton(ExecutionProperty.Key.DataStore));
+ super(UsedDataHandlingProperty.class, Collections.singleton(DataStoreProperty.class));
}
@Override
public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
dag.topologicalDo(irVertex ->
dag.getIncomingEdgesOf(irVertex).forEach(irEdge -> {
- if (irEdge.getProperty(ExecutionProperty.Key.UsedDataHandling) == null) {
- final DataStoreProperty.Value dataStoreValue = irEdge.getProperty(ExecutionProperty.Key.DataStore);
+ if (!irEdge.getPropertyValue(UsedDataHandlingProperty.class).isPresent()) {
+ final DataStoreProperty.Value dataStoreValue = irEdge.getPropertyValue(DataStoreProperty.class).get();
if (DataStoreProperty.Value.MemoryStore.equals(dataStoreValue)
|| DataStoreProperty.Value.SerializedMemoryStore.equals(dataStoreValue)) {
irEdge.setProperty(UsedDataHandlingProperty.of(UsedDataHandlingProperty.Value.Discard));
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPass.java
index 6b90111..d26320f 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPass.java
@@ -21,14 +21,11 @@ import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternPro
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.SourceVertex;
import edu.snu.nemo.common.dag.DAG;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import java.util.Collections;
import java.util.List;
-import static edu.snu.nemo.common.ir.executionproperty.ExecutionProperty.Key.DataCommunicationPattern;
-
/**
* Optimization pass for tagging parallelism execution property.
*/
@@ -52,7 +49,7 @@ public final class DefaultParallelismPass extends AnnotatingPass {
*/
public DefaultParallelismPass(final int desiredSourceParallelism,
final int shuffleDecreaseFactor) {
- super(ExecutionProperty.Key.Parallelism, Collections.singleton(DataCommunicationPattern));
+ super(ParallelismProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
this.desiredSourceParallelism = desiredSourceParallelism;
this.shuffleDecreaseFactor = shuffleDecreaseFactor;
}
@@ -68,7 +65,7 @@ public final class DefaultParallelismPass extends AnnotatingPass {
// After that, we set the parallelism as the number of split readers.
// (It can be more/less than the desired value.)
final SourceVertex sourceVertex = (SourceVertex) vertex;
- final Integer originalParallelism = vertex.getProperty(ExecutionProperty.Key.Parallelism);
+ final Integer originalParallelism = vertex.getPropertyValue(ParallelismProperty.class).get();
// We manipulate them if it is set as default value of 1.
if (originalParallelism.equals(1)) {
vertex.setProperty(ParallelismProperty.of(
@@ -78,14 +75,14 @@ public final class DefaultParallelismPass extends AnnotatingPass {
// No reason to propagate via Broadcast edges, as the data streams that will use the broadcasted data
// as a sideInput will have their own number of parallelism
final Integer o2oParallelism = inEdges.stream()
- .filter(edge -> DataCommunicationPatternProperty.Value.OneToOne
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))
- .mapToInt(edge -> edge.getSrc().getProperty(ExecutionProperty.Key.Parallelism))
+ .filter(edge -> DataCommunicationPatternProperty.Value.OneToOne
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get()))
+ .mapToInt(edge -> edge.getSrc().getPropertyValue(ParallelismProperty.class).get())
.max().orElse(1);
final Integer shuffleParallelism = inEdges.stream()
.filter(edge -> DataCommunicationPatternProperty.Value.Shuffle
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))
- .mapToInt(edge -> edge.getSrc().getProperty(ExecutionProperty.Key.Parallelism))
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get()))
+ .mapToInt(edge -> edge.getSrc().getPropertyValue(ParallelismProperty.class).get())
.map(i -> i / shuffleDecreaseFactor)
.max().orElse(1);
// We set the greater value as the parallelism.
@@ -93,7 +90,7 @@ public final class DefaultParallelismPass extends AnnotatingPass {
vertex.setProperty(ParallelismProperty.of(parallelism));
// synchronize one-to-one edges parallelism
recursivelySynchronizeO2OParallelism(dag, vertex, parallelism);
- } else if (vertex.getProperty(ExecutionProperty.Key.Parallelism) == null) {
+ } else if (!vertex.getPropertyValue(ParallelismProperty.class).isPresent()) {
throw new RuntimeException("There is a non-source vertex that doesn't have any inEdges "
+ "(excluding SideInput edges)");
} // No problem otherwise.
@@ -117,12 +114,12 @@ public final class DefaultParallelismPass extends AnnotatingPass {
final List<IREdge> inEdges = dag.getIncomingEdgesOf(vertex);
final Integer ancestorParallelism = inEdges.stream()
.filter(edge -> DataCommunicationPatternProperty.Value.OneToOne
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get()))
.map(IREdge::getSrc)
.mapToInt(inVertex -> recursivelySynchronizeO2OParallelism(dag, inVertex, parallelism))
.max().orElse(1);
final Integer maxParallelism = ancestorParallelism > parallelism ? ancestorParallelism : parallelism;
- final Integer myParallelism = vertex.getProperty(ExecutionProperty.Key.Parallelism);
+ final Integer myParallelism = vertex.getPropertyValue(ParallelismProperty.class).get();
// update the vertex with the max value.
if (maxParallelism > myParallelism) {
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultStagePartitioningPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultStagePartitioningPass.java
index e9c256c..7638e5d 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultStagePartitioningPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DefaultStagePartitioningPass.java
@@ -18,9 +18,12 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.PartitionerProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
import java.util.ArrayList;
@@ -42,12 +45,12 @@ public final class DefaultStagePartitioningPass extends AnnotatingPass {
* Default constructor.
*/
public DefaultStagePartitioningPass() {
- super(ExecutionProperty.Key.StageId, Stream.of(
- ExecutionProperty.Key.DataCommunicationPattern,
- ExecutionProperty.Key.ExecutorPlacement,
- ExecutionProperty.Key.DataFlowModel,
- ExecutionProperty.Key.Partitioner,
- ExecutionProperty.Key.Parallelism
+ super(StageIdProperty.class, Stream.of(
+ DataCommunicationPatternProperty.class,
+ ExecutorPlacementProperty.class,
+ DataFlowModelProperty.class,
+ PartitionerProperty.class,
+ ParallelismProperty.class
).collect(Collectors.toSet()));
}
@@ -85,17 +88,17 @@ public final class DefaultStagePartitioningPass extends AnnotatingPass {
// Filter candidate incoming edges that can be included in a stage with the vertex.
final Optional<List<IREdge>> inEdgesForStage = inEdgeList.map(e -> e.stream()
// One to one edges
- .filter(edge -> edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)
+ .filter(edge -> edge.getPropertyValue(DataCommunicationPatternProperty.class).get()
.equals(DataCommunicationPatternProperty.Value.OneToOne))
// MemoryStore placement
- .filter(edge -> edge.getProperty(ExecutionProperty.Key.DataStore)
+ .filter(edge -> edge.getPropertyValue(DataStoreProperty.class).get()
.equals(DataStoreProperty.Value.MemoryStore))
// if src and dst are placed on same container types
- .filter(edge -> edge.getSrc().getProperty(ExecutionProperty.Key.ExecutorPlacement)
- .equals(edge.getDst().getProperty(ExecutionProperty.Key.ExecutorPlacement)))
+ .filter(edge -> edge.getSrc().getPropertyValue(ExecutorPlacementProperty.class).get()
+ .equals(edge.getDst().getPropertyValue(ExecutorPlacementProperty.class).get()))
// if src and dst have same parallelism
- .filter(edge -> edge.getSrc().getProperty(ExecutionProperty.Key.Parallelism)
- .equals(edge.getDst().getProperty(ExecutionProperty.Key.Parallelism)))
+ .filter(edge -> edge.getSrc().getPropertyValue(ParallelismProperty.class).get()
+ .equals(edge.getDst().getPropertyValue(ParallelismProperty.class).get()))
// Src that is already included in a stage
.filter(edge -> vertexStageNumHashMap.containsKey(edge.getSrc()))
.collect(Collectors.toList()));
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DisaggregationEdgeDataStorePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DisaggregationEdgeDataStorePass.java
index 2b2f712..719de83 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DisaggregationEdgeDataStorePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DisaggregationEdgeDataStorePass.java
@@ -18,7 +18,6 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.dag.DAG;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
import java.util.Collections;
@@ -33,7 +32,7 @@ public final class DisaggregationEdgeDataStorePass extends AnnotatingPass {
* Default constructor.
*/
public DisaggregationEdgeDataStorePass() {
- super(ExecutionProperty.Key.DataStore, Collections.singleton(ExecutionProperty.Key.DataStore));
+ super(DataStoreProperty.class, Collections.singleton(DataStoreProperty.class));
}
@Override
@@ -42,7 +41,7 @@ public final class DisaggregationEdgeDataStorePass extends AnnotatingPass {
final List<IREdge> inEdges = dag.getIncomingEdgesOf(vertex);
inEdges.forEach(edge -> {
if (DataStoreProperty.Value.LocalFileStore
- .equals(edge.getProperty(ExecutionProperty.Key.DataStore))) {
+ .equals(edge.getPropertyValue(DataStoreProperty.class).get())) {
edge.setProperty(DataStoreProperty.of(DataStoreProperty.Value.GlusterFileStore));
}
});
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DuplicateEdgeGroupSizePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DuplicateEdgeGroupSizePass.java
index 06ab3d3..c4db802 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DuplicateEdgeGroupSizePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/DuplicateEdgeGroupSizePass.java
@@ -17,11 +17,12 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupPropertyValue;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import java.util.HashMap;
+import java.util.Optional;
/**
* A pass for annotate duplicate data for each edge.
@@ -32,7 +33,7 @@ public final class DuplicateEdgeGroupSizePass extends AnnotatingPass {
* Default constructor.
*/
public DuplicateEdgeGroupSizePass() {
- super(ExecutionProperty.Key.DuplicateEdgeGroup);
+ super(DuplicateEdgeGroupProperty.class);
}
@Override
@@ -40,10 +41,10 @@ public final class DuplicateEdgeGroupSizePass extends AnnotatingPass {
final HashMap<String, Integer> groupIdToGroupSize = new HashMap<>();
dag.topologicalDo(vertex -> dag.getIncomingEdgesOf(vertex)
.forEach(e -> {
- final DuplicateEdgeGroupPropertyValue duplicateEdgeGroupProperty =
- e.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- if (duplicateEdgeGroupProperty != null) {
- final String groupId = duplicateEdgeGroupProperty.getGroupId();
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateEdgeGroupProperty =
+ e.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ if (duplicateEdgeGroupProperty.isPresent()) {
+ final String groupId = duplicateEdgeGroupProperty.get().getGroupId();
final Integer currentCount = groupIdToGroupSize.getOrDefault(groupId, 0);
groupIdToGroupSize.put(groupId, currentCount + 1);
}
@@ -51,12 +52,12 @@ public final class DuplicateEdgeGroupSizePass extends AnnotatingPass {
dag.topologicalDo(vertex -> dag.getIncomingEdgesOf(vertex)
.forEach(e -> {
- final DuplicateEdgeGroupPropertyValue duplicateEdgeGroupProperty =
- e.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- if (duplicateEdgeGroupProperty != null) {
- final String groupId = duplicateEdgeGroupProperty.getGroupId();
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateEdgeGroupProperty =
+ e.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ if (duplicateEdgeGroupProperty.isPresent()) {
+ final String groupId = duplicateEdgeGroupProperty.get().getGroupId();
if (groupIdToGroupSize.containsKey(groupId)) {
- duplicateEdgeGroupProperty.setGroupSize(groupIdToGroupSize.get(groupId));
+ duplicateEdgeGroupProperty.get().setGroupSize(groupIdToGroupSize.get(groupId));
}
}
}));
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataFlowModelPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataFlowModelPass.java
index c110b3b..b74a995 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataFlowModelPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataFlowModelPass.java
@@ -18,12 +18,11 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
import static edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.PadoEdgeDataStorePass.fromTransientToReserved;
@@ -35,9 +34,7 @@ public final class PadoEdgeDataFlowModelPass extends AnnotatingPass {
* Default constructor.
*/
public PadoEdgeDataFlowModelPass() {
- super(ExecutionProperty.Key.DataFlowModel, Stream.of(
- ExecutionProperty.Key.ExecutorPlacement
- ).collect(Collectors.toSet()));
+ super(DataFlowModelProperty.class, Collections.singleton(ExecutorPlacementProperty.class));
}
@Override
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataStorePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataStorePass.java
index e0ca239..0777c1c 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataStorePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoEdgeDataStorePass.java
@@ -19,13 +19,11 @@ import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.dag.DAG;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
/**
* Pado pass for tagging edges with DataStore ExecutionProperty.
@@ -35,9 +33,7 @@ public final class PadoEdgeDataStorePass extends AnnotatingPass {
* Default constructor.
*/
public PadoEdgeDataStorePass() {
- super(ExecutionProperty.Key.DataStore, Stream.of(
- ExecutionProperty.Key.ExecutorPlacement
- ).collect(Collectors.toSet()));
+ super(DataStoreProperty.class, Collections.singleton(ExecutorPlacementProperty.class));
}
@Override
@@ -49,7 +45,7 @@ public final class PadoEdgeDataStorePass extends AnnotatingPass {
if (fromTransientToReserved(edge) || fromReservedToTransient(edge)) {
edge.setProperty(DataStoreProperty.of(DataStoreProperty.Value.LocalFileStore));
} else if (DataCommunicationPatternProperty.Value.OneToOne
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
edge.setProperty(DataStoreProperty.of(DataStoreProperty.Value.MemoryStore));
} else {
edge.setProperty(DataStoreProperty.of(DataStoreProperty.Value.LocalFileStore));
@@ -67,9 +63,9 @@ public final class PadoEdgeDataStorePass extends AnnotatingPass {
*/
static boolean fromTransientToReserved(final IREdge irEdge) {
return ExecutorPlacementProperty.TRANSIENT
- .equals(irEdge.getSrc().getProperty(ExecutionProperty.Key.ExecutorPlacement))
+ .equals(irEdge.getSrc().getPropertyValue(ExecutorPlacementProperty.class).get())
&& ExecutorPlacementProperty.RESERVED
- .equals(irEdge.getDst().getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ .equals(irEdge.getDst().getPropertyValue(ExecutorPlacementProperty.class).get());
}
/**
@@ -79,8 +75,8 @@ public final class PadoEdgeDataStorePass extends AnnotatingPass {
*/
static boolean fromReservedToTransient(final IREdge irEdge) {
return ExecutorPlacementProperty.RESERVED
- .equals(irEdge.getSrc().getProperty(ExecutionProperty.Key.ExecutorPlacement))
+ .equals(irEdge.getSrc().getPropertyValue(ExecutorPlacementProperty.class).get())
&& ExecutorPlacementProperty.TRANSIENT
- .equals(irEdge.getDst().getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ .equals(irEdge.getDst().getPropertyValue(ExecutorPlacementProperty.class).get());
}
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoVertexExecutorPlacementPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoVertexExecutorPlacementPass.java
index 431f692..64e4dfb 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoVertexExecutorPlacementPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/PadoVertexExecutorPlacementPass.java
@@ -19,12 +19,10 @@ import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.dag.DAG;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
/**
* Pado pass for tagging vertices.
@@ -34,9 +32,7 @@ public final class PadoVertexExecutorPlacementPass extends AnnotatingPass {
* Default constructor.
*/
public PadoVertexExecutorPlacementPass() {
- super(ExecutionProperty.Key.ExecutorPlacement, Stream.of(
- ExecutionProperty.Key.DataCommunicationPattern
- ).collect(Collectors.toSet()));
+ super(ExecutorPlacementProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -63,7 +59,7 @@ public final class PadoVertexExecutorPlacementPass extends AnnotatingPass {
*/
private boolean hasM2M(final List<IREdge> irEdges) {
return irEdges.stream().anyMatch(edge ->
- edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)
+ edge.getPropertyValue(DataCommunicationPatternProperty.class).get()
.equals(DataCommunicationPatternProperty.Value.Shuffle));
}
@@ -75,8 +71,8 @@ public final class PadoVertexExecutorPlacementPass extends AnnotatingPass {
private boolean allO2OFromReserved(final List<IREdge> irEdges) {
return irEdges.stream()
.allMatch(edge -> DataCommunicationPatternProperty.Value.OneToOne.equals(
- edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))
- && edge.getSrc().getProperty(ExecutionProperty.Key.ExecutorPlacement).equals(
+ edge.getPropertyValue(DataCommunicationPatternProperty.class).get())
+ && edge.getSrc().getPropertyValue(ExecutorPlacementProperty.class).get().equals(
ExecutorPlacementProperty.RESERVED));
}
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ReviseInterStageEdgeDataStorePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ReviseInterStageEdgeDataStorePass.java
index 0e066c4..35c8c5a 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ReviseInterStageEdgeDataStorePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ReviseInterStageEdgeDataStorePass.java
@@ -18,12 +18,11 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
+import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
/**
* Edge data store pass to process inter-stage memory store edges.
@@ -33,9 +32,7 @@ public final class ReviseInterStageEdgeDataStorePass extends AnnotatingPass {
* Default constructor.
*/
public ReviseInterStageEdgeDataStorePass() {
- super(ExecutionProperty.Key.DataStore, Stream.of(
- ExecutionProperty.Key.StageId
- ).collect(Collectors.toSet()));
+ super(DataStoreProperty.class, Collections.singleton(StageIdProperty.class));
}
@Override
@@ -44,9 +41,9 @@ public final class ReviseInterStageEdgeDataStorePass extends AnnotatingPass {
final List<IREdge> inEdges = dag.getIncomingEdgesOf(vertex);
if (!inEdges.isEmpty()) {
inEdges.forEach(edge -> {
- if (DataStoreProperty.Value.MemoryStore.equals(edge.getProperty(ExecutionProperty.Key.DataStore))
- && !edge.getSrc().getProperty(ExecutionProperty.Key.StageId)
- .equals(edge.getDst().getProperty(ExecutionProperty.Key.StageId))) {
+ if (DataStoreProperty.Value.MemoryStore.equals(edge.getPropertyValue(DataStoreProperty.class).get())
+ && !edge.getSrc().getPropertyValue(StageIdProperty.class).get()
+ .equals(edge.getDst().getPropertyValue(StageIdProperty.class).get())) {
edge.setProperty(DataStoreProperty.of(DataStoreProperty.Value.LocalFileStore));
}
});
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataFlowModelPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataFlowModelPass.java
index 8ff037b..95154b4 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataFlowModelPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataFlowModelPass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import java.util.Collections;
@@ -34,7 +33,7 @@ public final class SailfishEdgeDataFlowModelPass extends AnnotatingPass {
* Default constructor.
*/
public SailfishEdgeDataFlowModelPass() {
- super(ExecutionProperty.Key.DataFlowModel, Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(DataFlowModelProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -42,7 +41,7 @@ public final class SailfishEdgeDataFlowModelPass extends AnnotatingPass {
dag.getVertices().forEach(vertex -> {
final List<IREdge> inEdges = dag.getIncomingEdgesOf(vertex);
inEdges.forEach(edge -> {
- if (edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)
+ if (edge.getPropertyValue(DataCommunicationPatternProperty.class).get()
.equals(DataCommunicationPatternProperty.Value.Shuffle)) {
edge.setProperty(DataFlowModelProperty.of(DataFlowModelProperty.Value.Push)); // Push to the merger vertex.
} else {
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataStorePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataStorePass.java
index fe5add0..6fae5ef 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataStorePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeDataStorePass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
import java.util.Collections;
@@ -33,7 +32,7 @@ public final class SailfishEdgeDataStorePass extends AnnotatingPass {
* Default constructor.
*/
public SailfishEdgeDataStorePass() {
- super(ExecutionProperty.Key.DataStore, Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(DataStoreProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -42,10 +41,10 @@ public final class SailfishEdgeDataStorePass extends AnnotatingPass {
// Find the merger vertex inserted by reshaping pass.
if (dag.getIncomingEdgesOf(vertex).stream().anyMatch(irEdge ->
DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))) {
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get()))) {
dag.getIncomingEdgesOf(vertex).forEach(edgeToMerger -> {
if (DataCommunicationPatternProperty.Value.Shuffle
- .equals(edgeToMerger.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(edgeToMerger.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
// Pass data through memory to the merger vertex.
edgeToMerger.setProperty(DataStoreProperty.of(DataStoreProperty.Value.SerializedMemoryStore));
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeUsedDataHandlingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeUsedDataHandlingPass.java
index 9e69f7f..dd3a532 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeUsedDataHandlingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/SailfishEdgeUsedDataHandlingPass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.UsedDataHandlingProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import java.util.Collections;
@@ -34,14 +33,14 @@ public final class SailfishEdgeUsedDataHandlingPass extends AnnotatingPass {
* Default constructor.
*/
public SailfishEdgeUsedDataHandlingPass() {
- super(ExecutionProperty.Key.UsedDataHandling, Collections.singleton(ExecutionProperty.Key.DataFlowModel));
+ super(UsedDataHandlingProperty.class, Collections.singleton(DataFlowModelProperty.class));
}
@Override
public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
dag.topologicalDo(irVertex ->
dag.getIncomingEdgesOf(irVertex).forEach(irEdge -> {
- final DataFlowModelProperty.Value dataFlowModel = irEdge.getProperty(ExecutionProperty.Key.DataFlowModel);
+ final DataFlowModelProperty.Value dataFlowModel = irEdge.getPropertyValue(DataFlowModelProperty.class).get();
if (DataFlowModelProperty.Value.Push.equals(dataFlowModel)) {
irEdge.setProperty(UsedDataHandlingProperty.of(UsedDataHandlingProperty.Value.Discard));
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPass.java
index 6a48470..e76f4ba 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPass.java
@@ -18,18 +18,19 @@ package edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating;
import com.google.common.collect.Lists;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.PartitionerProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.ScheduleGroupIndexProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static edu.snu.nemo.common.ir.executionproperty.ExecutionProperty.Key.DataFlowModel;
-import static edu.snu.nemo.common.ir.executionproperty.ExecutionProperty.Key.StageId;
-
/**
* A pass for assigning each stages in schedule groups.
* We traverse the DAG topologically to find the dependency information between stages and number them appropriately
@@ -40,13 +41,13 @@ public final class ScheduleGroupPass extends AnnotatingPass {
* Default constructor.
*/
public ScheduleGroupPass() {
- super(ExecutionProperty.Key.ScheduleGroupIndex, Stream.of(
- ExecutionProperty.Key.StageId,
- ExecutionProperty.Key.DataCommunicationPattern,
- ExecutionProperty.Key.ExecutorPlacement,
- ExecutionProperty.Key.DataFlowModel,
- ExecutionProperty.Key.Partitioner,
- ExecutionProperty.Key.Parallelism
+ super(ScheduleGroupIndexProperty.class, Stream.of(
+ StageIdProperty.class,
+ DataCommunicationPatternProperty.class,
+ ExecutorPlacementProperty.class,
+ DataFlowModelProperty.class,
+ PartitionerProperty.class,
+ ParallelismProperty.class
).collect(Collectors.toSet()));
}
@@ -55,7 +56,8 @@ public final class ScheduleGroupPass extends AnnotatingPass {
@Override
public DAG<IRVertex, IREdge> apply(final DAG<IRVertex, IREdge> dag) {
// We assume that the input dag is tagged with stage ids.
- if (dag.getVertices().stream().anyMatch(irVertex -> irVertex.getProperty(StageId) == null)) {
+ if (dag.getVertices().stream()
+ .anyMatch(irVertex -> !irVertex.getPropertyValue(StageIdProperty.class).isPresent())) {
throw new RuntimeException("There exists an IR vertex going through ScheduleGroupPass "
+ "without stage id tagged.");
}
@@ -63,12 +65,12 @@ public final class ScheduleGroupPass extends AnnotatingPass {
// Map of stage id to the stage ids that it depends on.
final Map<Integer, Set<Integer>> dependentStagesMap = new HashMap<>();
dag.topologicalDo(irVertex -> {
- final Integer currentStageId = irVertex.getProperty(StageId);
+ final Integer currentStageId = irVertex.getPropertyValue(StageIdProperty.class).get();
dependentStagesMap.putIfAbsent(currentStageId, new HashSet<>());
// while traversing, we find the stages that point to the current stage and add them to the list.
dag.getIncomingEdgesOf(irVertex).stream()
.map(IREdge::getSrc)
- .mapToInt(vertex -> vertex.getProperty(StageId))
+ .mapToInt(vertex -> vertex.getPropertyValue(StageIdProperty.class).get())
.filter(n -> n != currentStageId)
.forEach(n -> dependentStagesMap.get(currentStageId).add(n));
});
@@ -107,17 +109,19 @@ public final class ScheduleGroupPass extends AnnotatingPass {
Lists.reverse(dag.getTopologicalSort()).forEach(v -> {
// get the destination vertices of the edges that are marked as push
final List<IRVertex> pushConnectedVertices = dag.getOutgoingEdgesOf(v).stream()
- .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getProperty(DataFlowModel)))
+ .filter(e -> DataFlowModelProperty.Value.Push.equals(e.getPropertyValue(DataFlowModelProperty.class).get()))
.map(IREdge::getDst)
.collect(Collectors.toList());
if (!pushConnectedVertices.isEmpty()) { // if we need to do something,
// we find the min value of the destination schedule groups.
final Integer newSchedulerGroupIndex = pushConnectedVertices.stream()
- .mapToInt(irVertex -> stageIdToScheduleGroupIndexMap.get(irVertex.<Integer>getProperty(StageId)))
+ .mapToInt(irVertex -> stageIdToScheduleGroupIndexMap
+ .get(irVertex.getPropertyValue(StageIdProperty.class).get()))
.min().orElseThrow(() -> new RuntimeException("a list was not empty, but produced an empty result"));
// overwrite
- final Integer originalScheduleGroupIndex = stageIdToScheduleGroupIndexMap.get(v.<Integer>getProperty(StageId));
- stageIdToScheduleGroupIndexMap.replace(v.getProperty(StageId), newSchedulerGroupIndex);
+ final Integer originalScheduleGroupIndex = stageIdToScheduleGroupIndexMap
+ .get(v.getPropertyValue(StageIdProperty.class).get());
+ stageIdToScheduleGroupIndexMap.replace(v.getPropertyValue(StageIdProperty.class).get(), newSchedulerGroupIndex);
// shift those if it came too far
if (stageIdToScheduleGroupIndexMap.values().stream()
.noneMatch(stageIndex -> stageIndex.equals(originalScheduleGroupIndex))) { // if it doesn't exist
@@ -133,9 +137,8 @@ public final class ScheduleGroupPass extends AnnotatingPass {
});
// do the tagging
- dag.topologicalDo(irVertex ->
- irVertex.setProperty(
- ScheduleGroupIndexProperty.of(stageIdToScheduleGroupIndexMap.get(irVertex.<Integer>getProperty(StageId)))));
+ dag.topologicalDo(irVertex -> irVertex.setProperty(ScheduleGroupIndexProperty.of(
+ stageIdToScheduleGroupIndexMap.get(irVertex.getPropertyValue(StageIdProperty.class).get()))));
return dag;
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ShuffleEdgePushPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ShuffleEdgePushPass.java
index 4bd3117..d548a1f 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ShuffleEdgePushPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/annotating/ShuffleEdgePushPass.java
@@ -19,7 +19,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import java.util.Collections;
@@ -34,7 +33,7 @@ public final class ShuffleEdgePushPass extends AnnotatingPass {
* Default constructor.
*/
public ShuffleEdgePushPass() {
- super(ExecutionProperty.Key.DataFlowModel, Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(DataFlowModelProperty.class, Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -43,7 +42,7 @@ public final class ShuffleEdgePushPass extends AnnotatingPass {
final List<IREdge> inEdges = dag.getIncomingEdgesOf(vertex);
if (!inEdges.isEmpty()) {
inEdges.forEach(edge -> {
- if (edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)
+ if (edge.getPropertyValue(DataCommunicationPatternProperty.class).get()
.equals(DataCommunicationPatternProperty.Value.Shuffle)) {
edge.setProperty(DataFlowModelProperty.of(DataFlowModelProperty.Value.Push));
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/composite/CompositePass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/composite/CompositePass.java
index e6347c7..e5628cc 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/composite/CompositePass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/composite/CompositePass.java
@@ -29,7 +29,7 @@ import java.util.*;
*/
public abstract class CompositePass implements CompileTimePass {
private final List<CompileTimePass> passList;
- private final Set<ExecutionProperty.Key> prerequisiteExecutionProperties;
+ private final Set<Class<? extends ExecutionProperty>> prerequisiteExecutionProperties;
/**
* Constructor.
@@ -75,7 +75,7 @@ public abstract class CompositePass implements CompileTimePass {
}
@Override
- public final Set<ExecutionProperty.Key> getPrerequisiteExecutionProperties() {
+ public final Set<Class<? extends ExecutionProperty>> getPrerequisiteExecutionProperties() {
return prerequisiteExecutionProperties;
}
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/CommonSubexpressionEliminationPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/CommonSubexpressionEliminationPass.java
index 9777ebb..64b1060 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/CommonSubexpressionEliminationPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/CommonSubexpressionEliminationPass.java
@@ -15,14 +15,15 @@
*/
package edu.snu.nemo.compiler.optimizer.pass.compiletime.reshaping;
+import edu.snu.nemo.common.coder.Coder;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
import edu.snu.nemo.common.ir.vertex.transform.Transform;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import java.util.*;
import java.util.stream.Collectors;
@@ -38,7 +39,7 @@ public final class CommonSubexpressionEliminationPass extends ReshapingPass {
* Default constructor.
*/
public CommonSubexpressionEliminationPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -147,9 +148,12 @@ public final class CommonSubexpressionEliminationPass extends ReshapingPass {
final Set<IREdge> outListToModify = outEdges.get(ov);
outEdges.getOrDefault(ov, new HashSet<>()).forEach(e -> {
outListToModify.remove(e);
- final IREdge newIrEdge = new IREdge(e.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(e.getPropertyValue(DataCommunicationPatternProperty.class).get(),
operatorVertexToUse, e.getDst());
- newIrEdge.setProperty(CoderProperty.of(e.getProperty(ExecutionProperty.Key.Coder)));
+ final Optional<Coder> coderProperty = e.getPropertyValue(CoderProperty.class);
+ if (coderProperty.isPresent()) {
+ newIrEdge.setProperty(CoderProperty.of(coderProperty.get()));
+ }
outListToModify.add(newIrEdge);
});
outEdges.remove(ov);
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/DataSkewReshapingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/DataSkewReshapingPass.java
index d042584..4139264 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/DataSkewReshapingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/DataSkewReshapingPass.java
@@ -23,7 +23,6 @@ import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternPro
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import java.util.ArrayList;
import java.util.Collections;
@@ -42,7 +41,7 @@ public final class DataSkewReshapingPass extends ReshapingPass {
* Default constructor.
*/
public DataSkewReshapingPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -54,7 +53,7 @@ public final class DataSkewReshapingPass extends ReshapingPass {
// We care about OperatorVertices that have any incoming edges that are of type Shuffle.
if (v instanceof OperatorVertex && dag.getIncomingEdgesOf(v).stream().anyMatch(irEdge ->
DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))) {
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get()))) {
final MetricCollectionBarrierVertex<Integer, Long> metricCollectionBarrierVertex
= new MetricCollectionBarrierVertex<>();
metricCollectionVertices.add(metricCollectionBarrierVertex);
@@ -63,13 +62,13 @@ public final class DataSkewReshapingPass extends ReshapingPass {
dag.getIncomingEdgesOf(v).forEach(edge -> {
// we insert the metric collection vertex when we meet a shuffle edge
if (DataCommunicationPatternProperty.Value.Shuffle
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
// We then insert the dynamicOptimizationVertex between the vertex and incoming vertices.
final IREdge newEdge = new IREdge(DataCommunicationPatternProperty.Value.OneToOne,
edge.getSrc(), metricCollectionBarrierVertex);
- newEdge.setProperty(CoderProperty.of(edge.getProperty(ExecutionProperty.Key.Coder)));
+ newEdge.setProperty(CoderProperty.of(edge.getPropertyValue(CoderProperty.class).get()));
- final IREdge edgeToGbK = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge edgeToGbK = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
metricCollectionBarrierVertex, v, edge.isSideInput());
edge.copyExecutionPropertiesTo(edgeToGbK);
builder.connectVertices(newEdge);
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopExtractionPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopExtractionPass.java
index bbcc430..c686eca 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopExtractionPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopExtractionPass.java
@@ -16,10 +16,10 @@
package edu.snu.nemo.compiler.optimizer.pass.compiletime.reshaping;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.LoopVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
import edu.snu.nemo.common.ir.vertex.SourceVertex;
@@ -37,7 +37,7 @@ public final class LoopExtractionPass extends ReshapingPass {
* Default constructor.
*/
public LoopExtractionPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -99,7 +99,7 @@ public final class LoopExtractionPass extends ReshapingPass {
final LoopVertex srcLoopVertex = dag.getAssignedLoopVertexOf(irEdge.getSrc());
srcLoopVertex.addDagOutgoingEdge(irEdge);
final IREdge edgeFromLoop =
- new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
srcLoopVertex, operatorVertex, irEdge.isSideInput());
irEdge.copyExecutionPropertiesTo(edgeFromLoop);
builder.connectVertices(edgeFromLoop);
@@ -147,7 +147,7 @@ public final class LoopExtractionPass extends ReshapingPass {
assignedLoopVertex.getBuilder().connectVertices(irEdge);
} else { // loop -> loop connection
assignedLoopVertex.addDagIncomingEdge(irEdge);
- final IREdge edgeToLoop = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge edgeToLoop = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
srcLoopVertex, assignedLoopVertex, irEdge.isSideInput());
irEdge.copyExecutionPropertiesTo(edgeToLoop);
builder.connectVertices(edgeToLoop);
@@ -155,7 +155,7 @@ public final class LoopExtractionPass extends ReshapingPass {
}
} else { // operator -> loop
assignedLoopVertex.addDagIncomingEdge(irEdge);
- final IREdge edgeToLoop = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge edgeToLoop = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
irEdge.getSrc(), assignedLoopVertex, irEdge.isSideInput());
irEdge.copyExecutionPropertiesTo(edgeToLoop);
builder.connectVertices(edgeToLoop);
@@ -226,13 +226,13 @@ public final class LoopExtractionPass extends ReshapingPass {
final IRVertex equivalentSrcVertex = equivalentVertices.get(srcVertex);
// add the new IREdge to the iterative incoming edges list.
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
equivalentSrcVertex, equivalentDstVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
finalRootLoopVertex.addIterativeIncomingEdge(newIrEdge);
} else {
// src is from outside the previous loop. vertex outside previous loop -> DAG.
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
srcVertex, equivalentDstVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
finalRootLoopVertex.addNonIterativeIncomingEdge(newIrEdge);
@@ -245,7 +245,7 @@ public final class LoopExtractionPass extends ReshapingPass {
final IRVertex dstVertex = edge.getDst();
final IRVertex equivalentSrcVertex = equivalentVertices.get(srcVertex);
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
equivalentSrcVertex, dstVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
finalRootLoopVertex.addDagOutgoingEdge(newIrEdge);
@@ -290,7 +290,7 @@ public final class LoopExtractionPass extends ReshapingPass {
if (edge.getSrc().equals(firstEquivalentVertex)) {
builder.connectVertices(edge);
} else {
- final IREdge newIrEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIrEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
firstEquivalentVertex, irVertex, edge.isSideInput());
edge.copyExecutionPropertiesTo(newIrEdge);
builder.connectVertices(newIrEdge);
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopOptimizations.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopOptimizations.java
index 79f605c..10d76e1 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopOptimizations.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/LoopOptimizations.java
@@ -16,12 +16,12 @@
package edu.snu.nemo.compiler.optimizer.pass.compiletime.reshaping;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.LoopVertex;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import java.util.*;
import java.util.function.IntPredicate;
@@ -102,7 +102,7 @@ public final class LoopOptimizations {
* Default constructor.
*/
public LoopFusionPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -159,8 +159,8 @@ public final class LoopOptimizations {
// inEdges.
inEdges.getOrDefault(loopVertex, new ArrayList<>()).forEach(irEdge -> {
if (builder.contains(irEdge.getSrc())) {
- final IREdge newIREdge = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
- irEdge.getSrc(), newLoopVertex, irEdge.isSideInput());
+ final IREdge newIREdge = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class)
+ .get(), irEdge.getSrc(), newLoopVertex, irEdge.isSideInput());
irEdge.copyExecutionPropertiesTo(newIREdge);
builder.connectVertices(newIREdge);
}
@@ -168,8 +168,8 @@ public final class LoopOptimizations {
// outEdges.
outEdges.getOrDefault(loopVertex, new ArrayList<>()).forEach(irEdge -> {
if (builder.contains(irEdge.getDst())) {
- final IREdge newIREdge = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
- newLoopVertex, irEdge.getDst(), irEdge.isSideInput());
+ final IREdge newIREdge = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class)
+ .get(), newLoopVertex, irEdge.getDst(), irEdge.isSideInput());
irEdge.copyExecutionPropertiesTo(newIREdge);
builder.connectVertices(newIREdge);
}
@@ -250,7 +250,7 @@ public final class LoopOptimizations {
* Default constructor.
*/
public LoopInvariantCodeMotionPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -286,9 +286,9 @@ public final class LoopOptimizations {
candidate.getValue().stream().map(IREdge::getSrc).anyMatch(edgeSrc -> edgeSrc.equals(e.getSrc())))
.forEach(edge -> {
edgesToRemove.add(edge);
- final IREdge newEdge = new IREdge(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newEdge = new IREdge(edge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
candidate.getKey(), edge.getDst(), edge.isSideInput());
- newEdge.setProperty(CoderProperty.of(edge.getProperty(ExecutionProperty.Key.Coder)));
+ newEdge.setProperty(CoderProperty.of(edge.getPropertyValue(CoderProperty.class).get()));
edgesToAdd.add(newEdge);
});
final List<IREdge> listToModify = inEdges.getOrDefault(loopVertex, new ArrayList<>());
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/ReshapingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/ReshapingPass.java
index 7c02a99..3cb5a9a 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/ReshapingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/ReshapingPass.java
@@ -26,7 +26,7 @@ import java.util.Set;
* It is ensured by the compiler that no execution properties are modified by a ReshapingPass.
*/
public abstract class ReshapingPass implements CompileTimePass {
- private final Set<ExecutionProperty.Key> prerequisiteExecutionProperties;
+ private final Set<Class<? extends ExecutionProperty>> prerequisiteExecutionProperties;
/**
* Default constructor.
@@ -39,12 +39,12 @@ public abstract class ReshapingPass implements CompileTimePass {
* Constructor.
* @param prerequisiteExecutionProperties prerequisite of execution properties.
*/
- public ReshapingPass(final Set<ExecutionProperty.Key> prerequisiteExecutionProperties) {
+ public ReshapingPass(final Set<Class<? extends ExecutionProperty>> prerequisiteExecutionProperties) {
this.prerequisiteExecutionProperties = prerequisiteExecutionProperties;
}
@Override
- public final Set<ExecutionProperty.Key> getPrerequisiteExecutionProperties() {
+ public final Set<Class<? extends ExecutionProperty>> getPrerequisiteExecutionProperties() {
return prerequisiteExecutionProperties;
}
}
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/SailfishRelayReshapingPass.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/SailfishRelayReshapingPass.java
index 567f0fb..e899da9 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/SailfishRelayReshapingPass.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/pass/compiletime/reshaping/SailfishRelayReshapingPass.java
@@ -20,7 +20,6 @@ import edu.snu.nemo.common.dag.DAGBuilder;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.IREdge;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.OperatorVertex;
import edu.snu.nemo.common.ir.vertex.transform.RelayTransform;
@@ -39,7 +38,7 @@ public final class SailfishRelayReshapingPass extends ReshapingPass {
* Default constructor.
*/
public SailfishRelayReshapingPass() {
- super(Collections.singleton(ExecutionProperty.Key.DataCommunicationPattern));
+ super(Collections.singleton(DataCommunicationPatternProperty.class));
}
@Override
@@ -51,10 +50,10 @@ public final class SailfishRelayReshapingPass extends ReshapingPass {
// has Shuffle as data communication pattern.
if (v instanceof OperatorVertex && dag.getIncomingEdgesOf(v).stream().anyMatch(irEdge ->
DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))) {
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get()))) {
dag.getIncomingEdgesOf(v).forEach(edge -> {
if (DataCommunicationPatternProperty.Value.Shuffle
- .equals(edge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(edge.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
// Insert a merger vertex having transform that write received data immediately
// before the vertex receiving shuffled data.
final OperatorVertex iFileMergerVertex = new OperatorVertex(new RelayTransform());
@@ -64,7 +63,7 @@ public final class SailfishRelayReshapingPass extends ReshapingPass {
edge.copyExecutionPropertiesTo(newEdgeToMerger);
final IREdge newEdgeFromMerger = new IREdge(DataCommunicationPatternProperty.Value.OneToOne,
iFileMergerVertex, v);
- newEdgeFromMerger.setProperty(CoderProperty.of(edge.getProperty(ExecutionProperty.Key.Coder)));
+ newEdgeFromMerger.setProperty(CoderProperty.of(edge.getPropertyValue(CoderProperty.class).get()));
builder.connectVertices(newEdgeToMerger);
builder.connectVertices(newEdgeFromMerger);
} else {
diff --git a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/policy/PolicyBuilder.java b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/policy/PolicyBuilder.java
index 1c25289..7b40556 100644
--- a/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/policy/PolicyBuilder.java
+++ b/compiler/optimizer/src/main/java/edu/snu/nemo/compiler/optimizer/policy/PolicyBuilder.java
@@ -16,7 +16,13 @@
package edu.snu.nemo.compiler.optimizer.policy;
import edu.snu.nemo.common.exception.CompileTimeOptimizationException;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.PartitionerProperty;
import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.CompileTimePass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.AnnotatingPass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.composite.CompositePass;
@@ -33,8 +39,8 @@ import java.util.Set;
public final class PolicyBuilder {
private final List<CompileTimePass> compileTimePasses;
private final List<RuntimePass<?>> runtimePasses;
- private final Set<ExecutionProperty.Key> finalizedExecutionProperties;
- private final Set<ExecutionProperty.Key> annotatedExecutionProperties;
+ private final Set<Class<? extends ExecutionProperty>> finalizedExecutionProperties;
+ private final Set<Class<? extends ExecutionProperty>> annotatedExecutionProperties;
private final Boolean strictPrerequisiteCheckMode;
/**
@@ -56,13 +62,13 @@ public final class PolicyBuilder {
this.annotatedExecutionProperties = new HashSet<>();
this.strictPrerequisiteCheckMode = strictPrerequisiteCheckMode;
// DataCommunicationPattern is already set when creating the IREdge itself.
- annotatedExecutionProperties.add(ExecutionProperty.Key.DataCommunicationPattern);
+ annotatedExecutionProperties.add(DataCommunicationPatternProperty.class);
// Some default values are already annotated.
- annotatedExecutionProperties.add(ExecutionProperty.Key.ExecutorPlacement);
- annotatedExecutionProperties.add(ExecutionProperty.Key.Parallelism);
- annotatedExecutionProperties.add(ExecutionProperty.Key.DataFlowModel);
- annotatedExecutionProperties.add(ExecutionProperty.Key.DataStore);
- annotatedExecutionProperties.add(ExecutionProperty.Key.Partitioner);
+ annotatedExecutionProperties.add(ExecutorPlacementProperty.class);
+ annotatedExecutionProperties.add(ParallelismProperty.class);
+ annotatedExecutionProperties.add(DataFlowModelProperty.class);
+ annotatedExecutionProperties.add(DataStoreProperty.class);
+ annotatedExecutionProperties.add(PartitionerProperty.class);
}
/**
diff --git a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/optimizer/RuntimeOptimizer.java b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/optimizer/RuntimeOptimizer.java
index 93b7964..5d33304 100644
--- a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/optimizer/RuntimeOptimizer.java
+++ b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/optimizer/RuntimeOptimizer.java
@@ -17,7 +17,6 @@ package edu.snu.nemo.runtime.common.optimizer;
import edu.snu.nemo.common.Pair;
import edu.snu.nemo.common.ir.vertex.MetricCollectionBarrierVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.executionproperty.DynamicOptimizationProperty;
import edu.snu.nemo.runtime.common.optimizer.pass.runtime.DataSkewRuntimePass;
import edu.snu.nemo.runtime.common.plan.PhysicalPlan;
@@ -44,7 +43,7 @@ public final class RuntimeOptimizer {
final PhysicalPlan originalPlan,
final MetricCollectionBarrierVertex metricCollectionBarrierVertex) {
final DynamicOptimizationProperty.Value dynamicOptimizationType =
- metricCollectionBarrierVertex.getProperty(ExecutionProperty.Key.DynamicOptimizationType);
+ metricCollectionBarrierVertex.getPropertyValue(DynamicOptimizationProperty.class).get();
switch (dynamicOptimizationType) {
case DataSkewRuntimePass:
diff --git a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/PhysicalPlanGenerator.java b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/PhysicalPlanGenerator.java
index fae871a..a11997e 100644
--- a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/PhysicalPlanGenerator.java
+++ b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/PhysicalPlanGenerator.java
@@ -16,13 +16,17 @@
package edu.snu.nemo.runtime.common.plan;
import edu.snu.nemo.common.ir.Readable;
+import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupPropertyValue;
import edu.snu.nemo.common.ir.vertex.*;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ScheduleGroupIndexProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
import edu.snu.nemo.conf.JobConf;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
import edu.snu.nemo.common.ir.edge.IREdge;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.exception.IllegalVertexOperationException;
import edu.snu.nemo.common.exception.PhysicalPlanGenerationException;
import org.apache.reef.tang.annotations.Parameter;
@@ -82,10 +86,10 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
final Map<String, List<StageEdge>> edgeGroupToIrEdge = new HashMap<>();
dagOfStages.topologicalDo(irVertex -> dagOfStages.getIncomingEdgesOf(irVertex).forEach(e -> {
- final DuplicateEdgeGroupPropertyValue duplicateEdgeGroupProperty =
- e.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- if (duplicateEdgeGroupProperty != null) {
- final String duplicateGroupId = duplicateEdgeGroupProperty.getGroupId();
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateEdgeGroupProperty =
+ e.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ if (duplicateEdgeGroupProperty.isPresent()) {
+ final String duplicateGroupId = duplicateEdgeGroupProperty.get().getGroupId();
edgeGroupToIrEdge.computeIfAbsent(duplicateGroupId, k -> new ArrayList<>()).add(e);
}
}));
@@ -93,10 +97,10 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
edgeGroupToIrEdge.forEach((id, edges) -> {
final StageEdge representativeEdge = edges.get(0);
final DuplicateEdgeGroupPropertyValue representativeProperty =
- representativeEdge.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
+ representativeEdge.getPropertyValue(DuplicateEdgeGroupProperty.class).get();
edges.forEach(e -> {
final DuplicateEdgeGroupPropertyValue duplicateEdgeGroupProperty =
- e.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
+ e.getPropertyValue(DuplicateEdgeGroupProperty.class).get();
duplicateEdgeGroupProperty.setRepresentativeEdgeId(representativeEdge.getId());
duplicateEdgeGroupProperty.setGroupSize(representativeProperty.getGroupSize());
});
@@ -114,7 +118,7 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
final Map<Integer, List<IRVertex>> vertexListForEachStage = new LinkedHashMap<>();
irDAG.topologicalDo(irVertex -> {
- final Integer stageNum = irVertex.getProperty(ExecutionProperty.Key.StageId);
+ final Integer stageNum = irVertex.getPropertyValue(StageIdProperty.class).get();
if (!vertexListForEachStage.containsKey(stageNum)) {
vertexListForEachStage.put(stageNum, new ArrayList<>());
}
@@ -133,13 +137,13 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
final IRVertex irVertexOfNewStage = stageVertices.stream().findAny()
.orElseThrow(() -> new RuntimeException("Error: List " + stageVertices.getClass() + " is Empty"));
final StageBuilder stageBuilder = new StageBuilder(
- irVertexOfNewStage.getProperty(ExecutionProperty.Key.StageId),
- irVertexOfNewStage.getProperty(ExecutionProperty.Key.Parallelism),
- irVertexOfNewStage.getProperty(ExecutionProperty.Key.ScheduleGroupIndex),
- irVertexOfNewStage.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ irVertexOfNewStage.getPropertyValue(StageIdProperty.class).get(),
+ irVertexOfNewStage.getPropertyValue(ParallelismProperty.class).get(),
+ irVertexOfNewStage.getPropertyValue(ScheduleGroupIndexProperty.class).get(),
+ irVertexOfNewStage.getPropertyValue(ExecutorPlacementProperty.class).get());
// Prepare useful variables.
- final int stageParallelism = irVertexOfNewStage.getProperty(ExecutionProperty.Key.Parallelism);
+ final int stageParallelism = irVertexOfNewStage.getPropertyValue(ParallelismProperty.class).get();
final List<Map<String, Readable>> vertexIdToReadables = new ArrayList<>(stageParallelism);
for (int i = 0; i < stageParallelism; i++) {
vertexIdToReadables.add(new HashMap<>());
@@ -237,9 +241,9 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
*/
private void integrityCheck(final List<IRVertex> stageVertices) {
final IRVertex firstVertex = stageVertices.get(0);
- final String placement = firstVertex.getProperty(ExecutionProperty.Key.ExecutorPlacement);
- final int scheduleGroup = firstVertex.<Integer>getProperty(ExecutionProperty.Key.ScheduleGroupIndex);
- final int parallelism = firstVertex.<Integer>getProperty(ExecutionProperty.Key.Parallelism);
+ final String placement = firstVertex.getPropertyValue(ExecutorPlacementProperty.class).get();
+ final int scheduleGroup = firstVertex.getPropertyValue(ScheduleGroupIndexProperty.class).get();
+ final int parallelism = firstVertex.getPropertyValue(ParallelismProperty.class).get();
stageVertices.forEach(irVertex -> {
// Check vertex type.
@@ -251,9 +255,9 @@ public final class PhysicalPlanGenerator implements Function<DAG<IRVertex, IREdg
// Check execution properties.
if ((placement != null
- && !placement.equals(irVertex.<String>getProperty(ExecutionProperty.Key.ExecutorPlacement)))
- || scheduleGroup != irVertex.<Integer>getProperty(ExecutionProperty.Key.ScheduleGroupIndex)
- || parallelism != irVertex.<Integer>getProperty(ExecutionProperty.Key.Parallelism)) {
+ && !placement.equals(irVertex.getPropertyValue(ExecutorPlacementProperty.class).get()))
+ || scheduleGroup != irVertex.getPropertyValue(ScheduleGroupIndexProperty.class).get()
+ || parallelism != irVertex.getPropertyValue(ParallelismProperty.class).get()) {
throw new RuntimeException("Vertices of the same stage have different execution properties: "
+ irVertex.getId());
}
diff --git a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/RuntimeEdge.java b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/RuntimeEdge.java
index e1a11ac..e4bd7eb 100644
--- a/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/RuntimeEdge.java
+++ b/runtime/common/src/main/java/edu/snu/nemo/runtime/common/plan/RuntimeEdge.java
@@ -17,8 +17,11 @@ package edu.snu.nemo.runtime.common.plan;
import edu.snu.nemo.common.dag.Edge;
import edu.snu.nemo.common.dag.Vertex;
+import edu.snu.nemo.common.ir.executionproperty.EdgeExecutionProperty;
import edu.snu.nemo.common.ir.executionproperty.ExecutionPropertyMap;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+
+import java.io.Serializable;
+import java.util.Optional;
/**
* Represents the edge between vertices in a logical/physical plan in runtime.
@@ -54,7 +57,8 @@ public class RuntimeEdge<V extends Vertex> extends Edge<V> {
* @param executionPropertyKey key of the execution property.
* @return the execution property.
*/
- public final <T> T getProperty(final ExecutionProperty.Key executionPropertyKey) {
+ public final <T extends Serializable> Optional<T> getPropertyValue(
+ final Class<? extends EdgeExecutionProperty<T>> executionPropertyKey) {
return executionProperties.get(executionPropertyKey);
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/Executor.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/Executor.java
index b52a17f..c9d3cc2 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/Executor.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/Executor.java
@@ -17,7 +17,8 @@ package edu.snu.nemo.runtime.executor;
import com.google.protobuf.ByteString;
import edu.snu.nemo.common.dag.DAG;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.CompressionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.conf.JobConf;
import edu.snu.nemo.common.exception.IllegalMessageException;
@@ -107,12 +108,15 @@ public final class Executor {
new TaskStateManager(task, executorId, persistentConnectionToMasterMap, metricMessageSender);
task.getTaskIncomingEdges().forEach(e -> serializerManager.register(e.getId(),
- e.getProperty(ExecutionProperty.Key.Coder), e.getExecutionProperties()));
+ e.getPropertyValue(CoderProperty.class).get(), e.getPropertyValue(CompressionProperty.class)
+ .orElse(null)));
task.getTaskOutgoingEdges().forEach(e -> serializerManager.register(e.getId(),
- e.getProperty(ExecutionProperty.Key.Coder), e.getExecutionProperties()));
+ e.getPropertyValue(CoderProperty.class).get(), e.getPropertyValue(CompressionProperty.class).
+ orElse(null)));
irDag.getVertices().forEach(v -> {
irDag.getOutgoingEdgesOf(v).forEach(e -> serializerManager.register(e.getId(),
- e.getProperty(ExecutionProperty.Key.Coder), e.getExecutionProperties()));
+ e.getPropertyValue(CoderProperty.class).get(), e.getPropertyValue(CompressionProperty.class)
+ .orElse(null)));
});
new TaskExecutor(
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
index 9d3920a..46c4609 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
@@ -19,8 +19,6 @@ import edu.snu.nemo.runtime.executor.data.streamchainer.CompressionStreamChainer
import edu.snu.nemo.runtime.executor.data.streamchainer.StreamChainer;
import edu.snu.nemo.common.coder.Coder;
import edu.snu.nemo.common.ir.edge.executionproperty.CompressionProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionPropertyMap;
import edu.snu.nemo.runtime.executor.data.streamchainer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,11 +49,22 @@ public final class SerializerManager {
*
* @param runtimeEdgeId id of the runtime edge.
* @param coder the corresponding coder.
- * @param propertyMap ExecutionPropertyMap of runtime edge
+ */
+ public void register(final String runtimeEdgeId,
+ final Coder coder) {
+ register(runtimeEdgeId, coder, null);
+ }
+
+ /**
+ * Register a coder for runtime edge.
+ *
+ * @param runtimeEdgeId id of the runtime edge.
+ * @param coder the corresponding coder.
+ * @param compressionProperty compression property, or null not to enable compression
*/
public void register(final String runtimeEdgeId,
final Coder coder,
- final ExecutionPropertyMap propertyMap) {
+ final CompressionProperty.Value compressionProperty) {
LOG.debug("{} edge id registering to SerializerManager", runtimeEdgeId);
final Serializer serializer = new Serializer(coder, Collections.emptyList());
runtimeEdgeIdToSerializer.putIfAbsent(runtimeEdgeId, serializer);
@@ -63,7 +72,6 @@ public final class SerializerManager {
final List<StreamChainer> streamChainerList = new ArrayList<>();
// Compression chain
- CompressionProperty.Compression compressionProperty = propertyMap.get(ExecutionProperty.Key.Compression);
if (compressionProperty != null) {
LOG.debug("Adding {} compression chain for {}",
compressionProperty, runtimeEdgeId);
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/streamchainer/CompressionStreamChainer.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/streamchainer/CompressionStreamChainer.java
index d321879..c5c2e92 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/streamchainer/CompressionStreamChainer.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/streamchainer/CompressionStreamChainer.java
@@ -30,14 +30,14 @@ import java.util.zip.GZIPOutputStream;
* {@link StreamChainer} for applying compression.
*/
public class CompressionStreamChainer implements StreamChainer {
- private final CompressionProperty.Compression compression;
+ private final CompressionProperty.Value compression;
/**
* Constructor.
*
* @param compression compression method.
*/
- public CompressionStreamChainer(final CompressionProperty.Compression compression) {
+ public CompressionStreamChainer(final CompressionProperty.Value compression) {
this.compression = compression;
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
index 9129826..fdaecd5 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
@@ -18,9 +18,10 @@ package edu.snu.nemo.runtime.executor.datatransfer;
import com.google.common.annotations.VisibleForTesting;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DuplicateEdgeGroupPropertyValue;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.runtime.common.RuntimeIdGenerator;
import edu.snu.nemo.runtime.common.data.KeyRange;
import edu.snu.nemo.runtime.common.plan.RuntimeEdge;
@@ -68,15 +69,14 @@ public final class InputReader extends DataTransfer {
* @return the read data.
*/
public List<CompletableFuture<DataUtil.IteratorWithNumBytes>> read() {
- DataCommunicationPatternProperty.Value comValue =
- (DataCommunicationPatternProperty.Value)
- runtimeEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern);
+ final Optional<DataCommunicationPatternProperty.Value> comValue =
+ runtimeEdge.getPropertyValue(DataCommunicationPatternProperty.class);
- if (comValue.equals(DataCommunicationPatternProperty.Value.OneToOne)) {
+ if (comValue.get().equals(DataCommunicationPatternProperty.Value.OneToOne)) {
return Collections.singletonList(readOneToOne());
- } else if (comValue.equals(DataCommunicationPatternProperty.Value.BroadCast)) {
+ } else if (comValue.get().equals(DataCommunicationPatternProperty.Value.BroadCast)) {
return readBroadcast();
- } else if (comValue.equals(DataCommunicationPatternProperty.Value.Shuffle)) {
+ } else if (comValue.get().equals(DataCommunicationPatternProperty.Value.Shuffle)) {
// If the dynamic optimization which detects data skew is enabled, read the data in the assigned range.
// TODO #492: Modularize the data communication pattern.
return readDataInRange();
@@ -87,20 +87,18 @@ public final class InputReader extends DataTransfer {
private CompletableFuture<DataUtil.IteratorWithNumBytes> readOneToOne() {
final String blockId = getBlockId(dstTaskIndex);
- return blockManagerWorker.queryBlock(blockId, getId(),
- (DataStoreProperty.Value) runtimeEdge.getProperty(ExecutionProperty.Key.DataStore),
- HashRange.all());
+ final Optional<DataStoreProperty.Value> dataStoreProperty = runtimeEdge.getPropertyValue(DataStoreProperty.class);
+ return blockManagerWorker.queryBlock(blockId, getId(), dataStoreProperty.get(), HashRange.all());
}
private List<CompletableFuture<DataUtil.IteratorWithNumBytes>> readBroadcast() {
final int numSrcTasks = this.getSourceParallelism();
+ final Optional<DataStoreProperty.Value> dataStoreProperty = runtimeEdge.getPropertyValue(DataStoreProperty.class);
final List<CompletableFuture<DataUtil.IteratorWithNumBytes>> futures = new ArrayList<>();
for (int srcTaskIdx = 0; srcTaskIdx < numSrcTasks; srcTaskIdx++) {
final String blockId = getBlockId(srcTaskIdx);
- futures.add(blockManagerWorker.queryBlock(blockId, getId(),
- (DataStoreProperty.Value) runtimeEdge.getProperty(ExecutionProperty.Key.DataStore),
- HashRange.all()));
+ futures.add(blockManagerWorker.queryBlock(blockId, getId(), dataStoreProperty.get(), HashRange.all()));
}
return futures;
@@ -113,6 +111,7 @@ public final class InputReader extends DataTransfer {
*/
private List<CompletableFuture<DataUtil.IteratorWithNumBytes>> readDataInRange() {
assert (runtimeEdge instanceof StageEdge);
+ final Optional<DataStoreProperty.Value> dataStoreProperty = runtimeEdge.getPropertyValue(DataStoreProperty.class);
final KeyRange hashRangeToRead =
((StageEdge) runtimeEdge).getTaskIdxToKeyRange().get(dstTaskIndex);
if (hashRangeToRead == null) {
@@ -125,9 +124,7 @@ public final class InputReader extends DataTransfer {
for (int srcTaskIdx = 0; srcTaskIdx < numSrcTasks; srcTaskIdx++) {
final String blockId = getBlockId(srcTaskIdx);
futures.add(
- blockManagerWorker.queryBlock(blockId, getId(),
- (DataStoreProperty.Value) runtimeEdge.getProperty(ExecutionProperty.Key.DataStore),
- hashRangeToRead));
+ blockManagerWorker.queryBlock(blockId, getId(), dataStoreProperty.get(), hashRangeToRead));
}
return futures;
@@ -144,12 +141,12 @@ public final class InputReader extends DataTransfer {
* @return the block id
*/
private String getBlockId(final int taskIdx) {
- final DuplicateEdgeGroupPropertyValue duplicateDataProperty =
- (DuplicateEdgeGroupPropertyValue) runtimeEdge.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- if (duplicateDataProperty == null || duplicateDataProperty.getGroupSize() <= 1) {
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateDataProperty =
+ runtimeEdge.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ if (!duplicateDataProperty.isPresent() || duplicateDataProperty.get().getGroupSize() <= 1) {
return RuntimeIdGenerator.generateBlockId(getId(), taskIdx);
}
- final String duplicateEdgeId = duplicateDataProperty.getRepresentativeEdgeId();
+ final String duplicateEdgeId = duplicateDataProperty.get().getRepresentativeEdgeId();
return RuntimeIdGenerator.generateBlockId(duplicateEdgeId, taskIdx);
}
@@ -168,10 +165,10 @@ public final class InputReader extends DataTransfer {
*/
public int getSourceParallelism() {
if (DataCommunicationPatternProperty.Value.OneToOne
- .equals(runtimeEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(runtimeEdge.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
return 1;
} else {
- final Integer numSrcTasks = srcVertex.getProperty(ExecutionProperty.Key.Parallelism);
+ final Integer numSrcTasks = srcVertex.getPropertyValue(ParallelismProperty.class).get();
return numSrcTasks;
}
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
index 8f686d4..b0dd080 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
@@ -19,7 +19,7 @@ import edu.snu.nemo.common.KeyExtractor;
import edu.snu.nemo.common.exception.*;
import edu.snu.nemo.common.ir.edge.executionproperty.*;
import edu.snu.nemo.common.ir.vertex.IRVertex;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.runtime.common.RuntimeIdGenerator;
import edu.snu.nemo.runtime.common.plan.RuntimeEdge;
import edu.snu.nemo.runtime.executor.data.BlockManagerWorker;
@@ -65,22 +65,22 @@ public final class OutputWriter extends DataTransfer implements AutoCloseable {
this.srcVertexId = srcRuntimeVertexId;
this.dstIrVertex = dstIrVertex;
this.blockManagerWorker = blockManagerWorker;
- this.blockStoreValue = runtimeEdge.getProperty(ExecutionProperty.Key.DataStore);
+ this.blockStoreValue = runtimeEdge.getPropertyValue(DataStoreProperty.class).get();
// Setup partitioner
final int dstParallelism = getDstParallelism();
- final KeyExtractor keyExtractor = runtimeEdge.getProperty(ExecutionProperty.Key.KeyExtractor);
+ final Optional<KeyExtractor> keyExtractor = runtimeEdge.getPropertyValue(KeyExtractorProperty.class);
final PartitionerProperty.Value partitionerPropertyValue =
- runtimeEdge.getProperty(ExecutionProperty.Key.Partitioner);
+ runtimeEdge.getPropertyValue(PartitionerProperty.class).get();
switch (partitionerPropertyValue) {
case IntactPartitioner:
this.partitioner = new IntactPartitioner();
break;
case HashPartitioner:
- this.partitioner = new HashPartitioner(dstParallelism, keyExtractor);
+ this.partitioner = new HashPartitioner(dstParallelism, keyExtractor.get());
break;
case DataSkewHashPartitioner:
- this.partitioner = new DataSkewHashPartitioner(hashRangeMultiplier, dstParallelism, keyExtractor);
+ this.partitioner = new DataSkewHashPartitioner(hashRangeMultiplier, dstParallelism, keyExtractor.get());
break;
default:
throw new UnsupportedPartitionerException(
@@ -88,11 +88,11 @@ public final class OutputWriter extends DataTransfer implements AutoCloseable {
}
blockToWrite = blockManagerWorker.createBlock(blockId, blockStoreValue);
- final DuplicateEdgeGroupPropertyValue duplicateDataProperty =
- runtimeEdge.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- nonDummyBlock = duplicateDataProperty == null
- || duplicateDataProperty.getRepresentativeEdgeId().equals(runtimeEdge.getId())
- || duplicateDataProperty.getGroupSize() <= 1;
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateDataProperty =
+ runtimeEdge.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ nonDummyBlock = !duplicateDataProperty.isPresent()
+ || duplicateDataProperty.get().getRepresentativeEdgeId().equals(runtimeEdge.getId())
+ || duplicateDataProperty.get().getGroupSize() <= 1;
}
/**
@@ -113,13 +113,13 @@ public final class OutputWriter extends DataTransfer implements AutoCloseable {
public void close() {
// Commit block.
final UsedDataHandlingProperty.Value usedDataHandling =
- runtimeEdge.getProperty(ExecutionProperty.Key.UsedDataHandling);
- final DuplicateEdgeGroupPropertyValue duplicateDataProperty =
- runtimeEdge.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- final int multiplier = duplicateDataProperty == null ? 1 : duplicateDataProperty.getGroupSize();
+ runtimeEdge.getPropertyValue(UsedDataHandlingProperty.class).get();
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateDataProperty =
+ runtimeEdge.getPropertyValue(DuplicateEdgeGroupProperty.class);
+ final int multiplier = duplicateDataProperty.isPresent() ? duplicateDataProperty.get().getGroupSize() : 1;
- final boolean isDataSizeMetricCollectionEdge = MetricCollectionProperty.Value.DataSkewRuntimePass
- .equals(runtimeEdge.getProperty(ExecutionProperty.Key.MetricCollection));
+ final boolean isDataSizeMetricCollectionEdge = Optional.of(MetricCollectionProperty.Value.DataSkewRuntimePass)
+ .equals(runtimeEdge.getPropertyValue(MetricCollectionProperty.class));
final Optional<Map<Integer, Long>> partitionSizeMap = blockToWrite.commit();
// Return the total size of the committed block.
if (partitionSizeMap.isPresent()) {
@@ -155,7 +155,7 @@ public final class OutputWriter extends DataTransfer implements AutoCloseable {
*/
private int getDstParallelism() {
return DataCommunicationPatternProperty.Value.OneToOne.equals(
- runtimeEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))
- ? 1 : dstIrVertex.getProperty(ExecutionProperty.Key.Parallelism);
+ runtimeEdge.getPropertyValue(DataCommunicationPatternProperty.class).get())
+ ? 1 : dstIrVertex.getPropertyValue(ParallelismProperty.class).get();
}
}
diff --git a/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/data/BlockStoreTest.java b/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/data/BlockStoreTest.java
index 6bac935..48ee45e 100644
--- a/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/data/BlockStoreTest.java
+++ b/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/data/BlockStoreTest.java
@@ -73,7 +73,7 @@ public final class BlockStoreTest {
private static final String TMP_FILE_DIRECTORY = "./tmpFiles";
private static final Coder CODER = PairCoder.of(IntCoder.of(), IntCoder.of());
private static final Serializer SERIALIZER = new Serializer(CODER,
- Collections.singletonList(new CompressionStreamChainer(CompressionProperty.Compression.LZ4)));
+ Collections.singletonList(new CompressionStreamChainer(CompressionProperty.Value.LZ4)));
private static final SerializerManager serializerManager = mock(SerializerManager.class);
private BlockManagerMaster blockManagerMaster;
private LocalMessageDispatcher messageDispatcher;
diff --git a/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/datatransfer/DataTransferTest.java b/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/datatransfer/DataTransferTest.java
index a0252ca..012701c 100644
--- a/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/datatransfer/DataTransferTest.java
+++ b/runtime/executor/src/test/java/edu/snu/nemo/runtime/executor/datatransfer/DataTransferTest.java
@@ -20,7 +20,6 @@ import edu.snu.nemo.common.coder.PairCoder;
import edu.snu.nemo.common.eventhandler.PubSubEventHandlerWrapper;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.*;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.SourceVertex;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
@@ -405,9 +404,10 @@ public final class DataTransferTest {
edgeProperties.put(DataCommunicationPatternProperty.of(commPattern));
edgeProperties.put(PartitionerProperty.of(PartitionerProperty.Value.HashPartitioner));
edgeProperties.put(DuplicateEdgeGroupProperty.of(new DuplicateEdgeGroupPropertyValue("dummy")));
- final DuplicateEdgeGroupPropertyValue duplicateDataProperty = edgeProperties.get(ExecutionProperty.Key.DuplicateEdgeGroup);
- duplicateDataProperty.setRepresentativeEdgeId(edgeId);
- duplicateDataProperty.setGroupSize(2);
+ final Optional<DuplicateEdgeGroupPropertyValue> duplicateDataProperty
+ = edgeProperties.get(DuplicateEdgeGroupProperty.class);
+ duplicateDataProperty.get().setRepresentativeEdgeId(edgeId);
+ duplicateDataProperty.get().setGroupSize(2);
edgeProperties.put(DataStoreProperty.of(store));
edgeProperties.put(UsedDataHandlingProperty.of(UsedDataHandlingProperty.Value.Keep));
@@ -523,8 +523,8 @@ public final class DataTransferTest {
private Pair<IRVertex, IRVertex> setupVertices(final String edgeId,
final BlockManagerWorker sender,
final BlockManagerWorker receiver) {
- serializerManagers.get(sender).register(edgeId, CODER, new ExecutionPropertyMap(""));
- serializerManagers.get(receiver).register(edgeId, CODER, new ExecutionPropertyMap(""));
+ serializerManagers.get(sender).register(edgeId, CODER);
+ serializerManagers.get(receiver).register(edgeId, CODER);
// Src setup
final SourceVertex srcVertex = new EmptyComponents.EmptySourceVertex("Source");
@@ -543,10 +543,10 @@ public final class DataTransferTest {
final String edgeId2,
final BlockManagerWorker sender,
final BlockManagerWorker receiver) {
- serializerManagers.get(sender).register(edgeId, CODER, new ExecutionPropertyMap(""));
- serializerManagers.get(receiver).register(edgeId, CODER, new ExecutionPropertyMap(""));
- serializerManagers.get(sender).register(edgeId2, CODER, new ExecutionPropertyMap(""));
- serializerManagers.get(receiver).register(edgeId2, CODER, new ExecutionPropertyMap(""));
+ serializerManagers.get(sender).register(edgeId, CODER);
+ serializerManagers.get(receiver).register(edgeId, CODER);
+ serializerManagers.get(sender).register(edgeId2, CODER);
+ serializerManagers.get(receiver).register(edgeId2, CODER);
// Src setup
final SourceVertex srcVertex = new EmptyComponents.EmptySourceVertex("Source");
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPassTest.java
index 52e4df4..b403f7b 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultEdgeCoderPassTest.java
@@ -19,7 +19,7 @@ import edu.snu.nemo.client.JobLauncher;
import edu.snu.nemo.common.coder.Coder;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
+import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.AnnotatingPass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.DefaultEdgeCoderPass;
@@ -48,19 +48,19 @@ public class DefaultEdgeCoderPassTest {
@Test
public void testAnnotatingPass() {
final AnnotatingPass coderPass = new DefaultEdgeCoderPass();
- assertEquals(ExecutionProperty.Key.Coder, coderPass.getExecutionPropertyToModify());
+ assertEquals(CoderProperty.class, coderPass.getExecutionPropertyToModify());
}
@Test
public void testNotOverride() {
// Get the first coder from the compiled DAG
final Coder compiledCoder = compiledDAG
- .getOutgoingEdgesOf(compiledDAG.getTopologicalSort().get(0)).get(0).getProperty(ExecutionProperty.Key.Coder);
+ .getOutgoingEdgesOf(compiledDAG.getTopologicalSort().get(0)).get(0).getPropertyValue(CoderProperty.class).get();
final DAG<IRVertex, IREdge> processedDAG = new DefaultEdgeCoderPass().apply(compiledDAG);
// Get the first coder from the processed DAG
- final Coder processedCoder = processedDAG
- .getOutgoingEdgesOf(processedDAG.getTopologicalSort().get(0)).get(0).getProperty(ExecutionProperty.Key.Coder);
+ final Coder processedCoder = processedDAG.getOutgoingEdgesOf(processedDAG.getTopologicalSort().get(0))
+ .get(0).getPropertyValue(CoderProperty.class).get();
assertEquals(compiledCoder, processedCoder); // It must not be changed.
}
@@ -68,12 +68,12 @@ public class DefaultEdgeCoderPassTest {
public void testSetToDefault() throws Exception {
// Remove the first coder from the compiled DAG (to let our pass to set as default coder).
compiledDAG.getOutgoingEdgesOf(compiledDAG.getTopologicalSort().get(0))
- .get(0).getExecutionProperties().remove(ExecutionProperty.Key.Coder);
+ .get(0).getExecutionProperties().remove(CoderProperty.class);
final DAG<IRVertex, IREdge> processedDAG = new DefaultEdgeCoderPass().apply(compiledDAG);
// Check whether the pass set the empty coder to our default coder.
- final Coder processedCoder = processedDAG
- .getOutgoingEdgesOf(processedDAG.getTopologicalSort().get(0)).get(0).getProperty(ExecutionProperty.Key.Coder);
+ final Coder processedCoder = processedDAG.getOutgoingEdgesOf(processedDAG.getTopologicalSort().get(0))
+ .get(0).getPropertyValue(CoderProperty.class).get();
assertEquals(Coder.DUMMY_CODER, processedCoder);
}
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPassTest.java
index e1d1a05..105eeb8 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/DefaultParallelismPassTest.java
@@ -18,9 +18,9 @@ package edu.snu.nemo.tests.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.client.JobLauncher;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.SourceVertex;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ParallelismProperty;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.AnnotatingPass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.DefaultParallelismPass;
import edu.snu.nemo.tests.compiler.CompilerTestUtil;
@@ -48,25 +48,25 @@ public class DefaultParallelismPassTest {
@Test
public void testAnnotatingPass() {
final AnnotatingPass parallelismPass = new DefaultParallelismPass();
- assertEquals(ExecutionProperty.Key.Parallelism, parallelismPass.getExecutionPropertyToModify());
+ assertEquals(ParallelismProperty.class, parallelismPass.getExecutionPropertyToModify());
}
@Test
- public void testParallelismOne() throws Exception {
+ public void testParallelismOne() {
final DAG<IRVertex, IREdge> processedDAG = new DefaultParallelismPass().apply(compiledDAG);
processedDAG.getTopologicalSort().forEach(irVertex ->
- assertEquals(1, irVertex.<Integer>getProperty(ExecutionProperty.Key.Parallelism).longValue()));
+ assertEquals(1, irVertex.getPropertyValue(ParallelismProperty.class).get().longValue()));
}
@Test
- public void testParallelismTen() throws Exception {
+ public void testParallelismTen() {
final int desiredSourceParallelism = 10;
final DAG<IRVertex, IREdge> processedDAG = new DefaultParallelismPass(desiredSourceParallelism, 2).apply(compiledDAG);
processedDAG.getTopologicalSort().stream()
.filter(irVertex -> irVertex instanceof SourceVertex)
.forEach(irVertex -> assertEquals(desiredSourceParallelism,
- irVertex.<Integer>getProperty(ExecutionProperty.Key.Parallelism).longValue()));
+ irVertex.getPropertyValue(ParallelismProperty.class).get().longValue()));
}
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPassTest.java
index a6f0e3d..9e222fc 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/annotating/ScheduleGroupPassTest.java
@@ -18,8 +18,8 @@ package edu.snu.nemo.tests.compiler.optimizer.pass.compiletime.annotating;
import edu.snu.nemo.client.JobLauncher;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
+import edu.snu.nemo.common.ir.vertex.executionproperty.ScheduleGroupIndexProperty;
import edu.snu.nemo.compiler.optimizer.CompiletimeOptimizer;
import edu.snu.nemo.tests.compiler.optimizer.policy.TestPolicy;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.AnnotatingPass;
@@ -47,7 +47,7 @@ public final class ScheduleGroupPassTest {
@Test
public void testAnnotatingPass() {
final AnnotatingPass scheduleGroupPass = new ScheduleGroupPass();
- assertEquals(ExecutionProperty.Key.ScheduleGroupIndex, scheduleGroupPass.getExecutionPropertyToModify());
+ assertEquals(ScheduleGroupIndexProperty.class, scheduleGroupPass.getExecutionPropertyToModify());
}
/**
@@ -60,10 +60,9 @@ public final class ScheduleGroupPassTest {
new TestPolicy(), "");
for (final IRVertex irVertex : processedDAG.getTopologicalSort()) {
- assertTrue(irVertex.getProperty(ExecutionProperty.Key.ScheduleGroupIndex) != null);
- final Integer currentScheduleGroupIndex = irVertex.getProperty(ExecutionProperty.Key.ScheduleGroupIndex);
+ final Integer currentScheduleGroupIndex = irVertex.getPropertyValue(ScheduleGroupIndexProperty.class).get();
final Integer largestScheduleGroupIndexOfParent = processedDAG.getParents(irVertex.getId()).stream()
- .mapToInt(v -> v.getProperty(ExecutionProperty.Key.ScheduleGroupIndex))
+ .mapToInt(v -> v.getPropertyValue(ScheduleGroupIndexProperty.class).get())
.max().orElse(0);
assertTrue(currentScheduleGroupIndex >= largestScheduleGroupIndexOfParent);
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DataSkewCompositePassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DataSkewCompositePassTest.java
index b29ed41..50b3cdb 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DataSkewCompositePassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DataSkewCompositePassTest.java
@@ -36,6 +36,7 @@ import org.powermock.modules.junit4.PowerMockRunner;
import java.util.HashSet;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import static org.junit.Assert.assertEquals;
@@ -62,7 +63,7 @@ public class DataSkewCompositePassTest {
final CompositePass dataSkewPass = new DataSkewCompositePass();
assertEquals(NUM_OF_PASSES_IN_DATA_SKEW_PASS, dataSkewPass.getPassList().size());
- final Set<ExecutionProperty.Key> prerequisites = new HashSet<>();
+ final Set<Class<? extends ExecutionProperty>> prerequisites = new HashSet<>();
dataSkewPass.getPassList().forEach(compileTimePass ->
prerequisites.addAll(compileTimePass.getPrerequisiteExecutionProperties()));
dataSkewPass.getPassList().forEach(compileTimePass -> {
@@ -85,7 +86,7 @@ public class DataSkewCompositePassTest {
final Long numOfShuffleGatherEdges = mrDAG.getVertices().stream().filter(irVertex ->
mrDAG.getIncomingEdgesOf(irVertex).stream().anyMatch(irEdge ->
DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern))))
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get())))
.count();
final DAG<IRVertex, IREdge> processedDAG = new DataSkewCompositePass().apply(mrDAG);
@@ -93,14 +94,14 @@ public class DataSkewCompositePassTest {
processedDAG.getVertices().stream().map(processedDAG::getIncomingEdgesOf)
.flatMap(List::stream)
.filter(irEdge -> DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get()))
.map(IREdge::getSrc)
.forEach(irVertex -> assertTrue(irVertex instanceof MetricCollectionBarrierVertex));
processedDAG.getVertices().forEach(v -> processedDAG.getOutgoingEdgesOf(v).stream()
- .filter(e -> MetricCollectionProperty.Value.DataSkewRuntimePass
- .equals(e.getProperty(ExecutionProperty.Key.MetricCollection)))
- .forEach(e -> assertEquals(e.getProperty(ExecutionProperty.Key.Partitioner),
- PartitionerProperty.Value.DataSkewHashPartitioner)));
+ .filter(e -> Optional.of(MetricCollectionProperty.Value.DataSkewRuntimePass)
+ .equals(e.getPropertyValue(MetricCollectionProperty.class)))
+ .forEach(e -> assertEquals(PartitionerProperty.Value.DataSkewHashPartitioner,
+ e.getPropertyValue(PartitionerProperty.class).get())));
}
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DisaggregationPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DisaggregationPassTest.java
index b0f65eb..898f74e 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DisaggregationPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/DisaggregationPassTest.java
@@ -20,8 +20,8 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
+import edu.snu.nemo.common.ir.vertex.executionproperty.StageIdProperty;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.DefaultParallelismPass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.DefaultStagePartitioningPass;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.DisaggregationEdgeDataStorePass;
@@ -59,13 +59,13 @@ public class DisaggregationPassTest {
processedDAG.getTopologicalSort().forEach(irVertex -> {
processedDAG.getIncomingEdgesOf(irVertex).forEach(edgeToMerger -> {
if (DataCommunicationPatternProperty.Value.OneToOne
- .equals(edgeToMerger.getProperty(ExecutionProperty.Key.DataCommunicationPattern))
- && edgeToMerger.getSrc().getProperty(ExecutionProperty.Key.StageId)
- .equals(edgeToMerger.getDst().getProperty(ExecutionProperty.Key.StageId))) {
- assertEquals(DataStoreProperty.Value.MemoryStore, edgeToMerger.getProperty(ExecutionProperty.Key.DataStore));
+ .equals(edgeToMerger.getPropertyValue(DataCommunicationPatternProperty.class).get())
+ && edgeToMerger.getSrc().getPropertyValue(StageIdProperty.class).get()
+ .equals(edgeToMerger.getDst().getPropertyValue(StageIdProperty.class).get())) {
+ assertEquals(DataStoreProperty.Value.MemoryStore, edgeToMerger.getPropertyValue(DataStoreProperty.class).get());
} else {
assertEquals(DataStoreProperty.Value.GlusterFileStore,
- edgeToMerger.getProperty(ExecutionProperty.Key.DataStore));
+ edgeToMerger.getPropertyValue(DataStoreProperty.class).get());
}
});
});
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/PadoCompositePassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/PadoCompositePassTest.java
index 5f29ed2..73014e2 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/PadoCompositePassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/PadoCompositePassTest.java
@@ -20,7 +20,6 @@ import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.ir.edge.IREdge;
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.executionproperty.ExecutorPlacementProperty;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.annotating.PadoEdgeDataStorePass;
@@ -53,41 +52,41 @@ public class PadoCompositePassTest {
final DAG<IRVertex, IREdge> processedDAG = new PadoCompositePass().apply(compiledDAG);
final IRVertex vertex1 = processedDAG.getTopologicalSort().get(0);
- assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex1.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex1.getPropertyValue(ExecutorPlacementProperty.class).get());
final IRVertex vertex5 = processedDAG.getTopologicalSort().get(1);
- assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex5.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex5.getPropertyValue(ExecutorPlacementProperty.class).get());
processedDAG.getIncomingEdgesOf(vertex5).forEach(irEdge -> {
- assertEquals(DataStoreProperty.Value.MemoryStore, irEdge.getProperty(ExecutionProperty.Key.DataStore));
- assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataStoreProperty.Value.MemoryStore, irEdge.getPropertyValue(DataStoreProperty.class).get());
+ assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
final IRVertex vertex6 = processedDAG.getTopologicalSort().get(2);
- assertEquals(ExecutorPlacementProperty.RESERVED, vertex6.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.RESERVED, vertex6.getPropertyValue(ExecutorPlacementProperty.class).get());
processedDAG.getIncomingEdgesOf(vertex6).forEach(irEdge -> {
- assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getProperty(ExecutionProperty.Key.DataStore));
- assertEquals(DataFlowModelProperty.Value.Push, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getPropertyValue(DataStoreProperty.class).get());
+ assertEquals(DataFlowModelProperty.Value.Push, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
final IRVertex vertex4 = processedDAG.getTopologicalSort().get(6);
- assertEquals(ExecutorPlacementProperty.RESERVED, vertex4.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.RESERVED, vertex4.getPropertyValue(ExecutorPlacementProperty.class).get());
processedDAG.getIncomingEdgesOf(vertex4).forEach(irEdge -> {
- assertEquals(DataStoreProperty.Value.MemoryStore, irEdge.getProperty(ExecutionProperty.Key.DataStore));
- assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataStoreProperty.Value.MemoryStore, irEdge.getPropertyValue(DataStoreProperty.class).get());
+ assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
final IRVertex vertex12 = processedDAG.getTopologicalSort().get(10);
- assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex12.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.TRANSIENT, vertex12.getPropertyValue(ExecutorPlacementProperty.class).get());
processedDAG.getIncomingEdgesOf(vertex12).forEach(irEdge -> {
- assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getProperty(ExecutionProperty.Key.DataStore));
- assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getPropertyValue(DataStoreProperty.class).get());
+ assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
final IRVertex vertex14 = processedDAG.getTopologicalSort().get(12);
- assertEquals(ExecutorPlacementProperty.RESERVED, vertex14.getProperty(ExecutionProperty.Key.ExecutorPlacement));
+ assertEquals(ExecutorPlacementProperty.RESERVED, vertex14.getPropertyValue(ExecutorPlacementProperty.class).get());
processedDAG.getIncomingEdgesOf(vertex14).forEach(irEdge -> {
- assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getProperty(ExecutionProperty.Key.DataStore));
- assertEquals(DataFlowModelProperty.Value.Push, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataStoreProperty.Value.LocalFileStore, irEdge.getPropertyValue(DataStoreProperty.class).get());
+ assertEquals(DataFlowModelProperty.Value.Push, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
}
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/SailfishPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/SailfishPassTest.java
index 80a8311..cd456fe 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/SailfishPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/composite/SailfishPassTest.java
@@ -22,7 +22,6 @@ import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternPro
import edu.snu.nemo.common.ir.edge.executionproperty.DataFlowModelProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.DataStoreProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.UsedDataHandlingProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.composite.SailfishPass;
import edu.snu.nemo.tests.compiler.CompilerTestUtil;
@@ -48,40 +47,40 @@ public class SailfishPassTest {
}
@Test
- public void testSailfish() throws Exception {
+ public void testSailfish() {
final DAG<IRVertex, IREdge> processedDAG = new SailfishPass().apply(compiledDAG);
processedDAG.getTopologicalSort().forEach(irVertex -> {
if (processedDAG.getIncomingEdgesOf(irVertex).stream().anyMatch(irEdge ->
DataCommunicationPatternProperty.Value.Shuffle
- .equals(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern)))) {
+ .equals(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get()))) {
// Merger vertex
processedDAG.getIncomingEdgesOf(irVertex).forEach(edgeToMerger -> {
if (DataCommunicationPatternProperty.Value.Shuffle
- .equals(edgeToMerger.getProperty(ExecutionProperty.Key.DataCommunicationPattern))) {
+ .equals(edgeToMerger.getPropertyValue(DataCommunicationPatternProperty.class).get())) {
assertEquals(DataFlowModelProperty.Value.Push,
- edgeToMerger.getProperty(ExecutionProperty.Key.DataFlowModel));
+ edgeToMerger.getPropertyValue(DataFlowModelProperty.class).get());
assertEquals(UsedDataHandlingProperty.Value.Discard,
- edgeToMerger.getProperty(ExecutionProperty.Key.UsedDataHandling));
+ edgeToMerger.getPropertyValue(UsedDataHandlingProperty.class).get());
assertEquals(DataStoreProperty.Value.SerializedMemoryStore,
- edgeToMerger.getProperty(ExecutionProperty.Key.DataStore));
+ edgeToMerger.getPropertyValue(DataStoreProperty.class).get());
} else {
assertEquals(DataFlowModelProperty.Value.Pull,
- edgeToMerger.getProperty(ExecutionProperty.Key.DataFlowModel));
+ edgeToMerger.getPropertyValue(DataFlowModelProperty.class).get());
}
});
processedDAG.getOutgoingEdgesOf(irVertex).forEach(edgeFromMerger -> {
assertEquals(DataFlowModelProperty.Value.Pull,
- edgeFromMerger.getProperty(ExecutionProperty.Key.DataFlowModel));
+ edgeFromMerger.getPropertyValue(DataFlowModelProperty.class).get());
assertEquals(DataCommunicationPatternProperty.Value.OneToOne,
- edgeFromMerger.getProperty(ExecutionProperty.Key.DataCommunicationPattern));
+ edgeFromMerger.getPropertyValue(DataCommunicationPatternProperty.class).get());
assertEquals(DataStoreProperty.Value.LocalFileStore,
- edgeFromMerger.getProperty(ExecutionProperty.Key.DataStore));
+ edgeFromMerger.getPropertyValue(DataStoreProperty.class).get());
});
} else {
// Non merger vertex.
processedDAG.getIncomingEdgesOf(irVertex).forEach(irEdge -> {
- assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getProperty(ExecutionProperty.Key.DataFlowModel));
+ assertEquals(DataFlowModelProperty.Value.Pull, irEdge.getPropertyValue(DataFlowModelProperty.class).get());
});
}
});
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopFusionPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopFusionPassTest.java
index 302ed4f..8a6b1a8 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopFusionPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopFusionPassTest.java
@@ -19,8 +19,8 @@ import edu.snu.nemo.client.JobLauncher;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.LoopVertex;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.reshaping.LoopExtractionPass;
@@ -112,15 +112,15 @@ public class LoopFusionPassTest {
final LoopVertex loopVertexToFollow) {
builder.addVertex(loopVertexToFollow);
loopVertexToFollow.getIterativeIncomingEdges().values().forEach(irEdges -> irEdges.forEach(irEdge -> {
- final IREdge newIREdge = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIREdge = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
vertexToBeFollowed, loopVertexToFollow);
- newIREdge.setProperty(CoderProperty.of(irEdge.getProperty(ExecutionProperty.Key.Coder)));
+ newIREdge.setProperty(CoderProperty.of(irEdge.getPropertyValue(CoderProperty.class).get()));
builder.connectVertices(newIREdge);
}));
loopVertexToFollow.getNonIterativeIncomingEdges().values().forEach(irEdges -> irEdges.forEach(irEdge -> {
- final IREdge newIREdge = new IREdge(irEdge.getProperty(ExecutionProperty.Key.DataCommunicationPattern),
+ final IREdge newIREdge = new IREdge(irEdge.getPropertyValue(DataCommunicationPatternProperty.class).get(),
irEdge.getSrc(), loopVertexToFollow);
- newIREdge.setProperty(CoderProperty.of(irEdge.getProperty(ExecutionProperty.Key.Coder)));
+ newIREdge.setProperty(CoderProperty.of(irEdge.getPropertyValue(CoderProperty.class).get()));
builder.connectVertices(newIREdge);
}));
}
diff --git a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopInvariantCodeMotionPassTest.java b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopInvariantCodeMotionPassTest.java
index ca551b0..de6ff56 100644
--- a/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopInvariantCodeMotionPassTest.java
+++ b/tests/src/test/java/edu/snu/nemo/tests/compiler/optimizer/pass/compiletime/reshaping/LoopInvariantCodeMotionPassTest.java
@@ -19,8 +19,8 @@ import edu.snu.nemo.client.JobLauncher;
import edu.snu.nemo.common.dag.DAG;
import edu.snu.nemo.common.dag.DAGBuilder;
import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.DataCommunicationPatternProperty;
import edu.snu.nemo.common.ir.edge.executionproperty.CoderProperty;
-import edu.snu.nemo.common.ir.executionproperty.ExecutionProperty;
import edu.snu.nemo.common.ir.vertex.IRVertex;
import edu.snu.nemo.common.ir.vertex.LoopVertex;
import edu.snu.nemo.compiler.optimizer.pass.compiletime.reshaping.LoopExtractionPass;
@@ -90,9 +90,9 @@ public class LoopInvariantCodeMotionPassTest {
} else {
final Optional<IREdge> incomingEdge = newDAGIncomingEdge.stream().findFirst();
assertTrue(incomingEdge.isPresent());
- final IREdge newIREdge = new IREdge(incomingEdge.get().getProperty(
- ExecutionProperty.Key.DataCommunicationPattern), incomingEdge.get().getSrc(), alsLoop);
- newIREdge.setProperty(CoderProperty.of(incomingEdge.get().getProperty(ExecutionProperty.Key.Coder)));
+ final IREdge newIREdge = new IREdge(incomingEdge.get().getPropertyValue(
+ DataCommunicationPatternProperty.class).get(), incomingEdge.get().getSrc(), alsLoop);
+ newIREdge.setProperty(CoderProperty.of(incomingEdge.get().getPropertyValue(CoderProperty.class).get()));
builder.connectVertices(newIREdge);
}
});
--
To stop receiving notification emails like this one, please contact
sanha@apache.org.