[REEF-1196] Allow specifying runtime name in evaluator request
This changeset addresses the issue by:
1)Adding runtime name to the evaluator request
2)Propagating runtime name to the evaluatiormanager
3)Adding runtime name to the Resource Events implementations
JIRA:
[REEF-1196](https://issues.apache.org/jira/browse/REEF-1196)
Pull Request:
Closes #848
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/driver/evaluator/EvaluatorRequest.java b/lang/java/reef-common/src/main/java/org/apache/reef/driver/evaluator/EvaluatorRequest.java
index 939d301..1f48f3a 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/driver/evaluator/EvaluatorRequest.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/driver/evaluator/EvaluatorRequest.java
@@ -38,17 +38,28 @@
private final int cores;
private final List<String> nodeNames;
private final List<String> rackNames;
+ private final String runtimeName;
EvaluatorRequest(final int number,
- final int megaBytes,
- final int cores,
- final List<String> nodeNames,
- final List<String> rackNames) {
+ final int megaBytes,
+ final int cores,
+ final List<String> nodeNames,
+ final List<String> rackNames) {
+ this(number, megaBytes, cores, nodeNames, rackNames, "");
+ }
+
+ EvaluatorRequest(final int number,
+ final int megaBytes,
+ final int cores,
+ final List<String> nodeNames,
+ final List<String> rackNames,
+ final String runtimeName) {
this.number = number;
this.megaBytes = megaBytes;
this.cores = cores;
this.nodeNames = nodeNames;
this.rackNames = rackNames;
+ this.runtimeName = runtimeName;
}
/**
@@ -116,6 +127,15 @@
}
/**
+ * Access the required runtime name.
+ *
+ * @return the runtime name that we need the Evaluator to run on
+ */
+ public String getRuntimeName() {
+ return runtimeName;
+ }
+
+ /**
* {@link EvaluatorRequest}s are build using this Builder.
*/
public static final class Builder implements org.apache.reef.util.Builder<EvaluatorRequest> {
@@ -125,6 +145,7 @@
private int cores = 1; //if not set, default to 1
private final List<String> nodeNames = new ArrayList<>();
private final List<String> rackNames = new ArrayList<>();
+ private String runtimeName = "";
private Builder() {
}
@@ -139,6 +160,7 @@
setNumber(request.getNumber());
setMemory(request.getMegaBytes());
setNumberOfCores(request.getNumberOfCores());
+ setRuntimeName(request.getRuntimeName());
for (final String nodeName : request.getNodeNames()) {
addNodeName(nodeName);
}
@@ -160,6 +182,18 @@
}
/**
+ * Set the name of the desired runtime.
+ *
+ * @param runtimeName to request for the Evaluator.
+ * @return this builder
+ */
+ @SuppressWarnings("checkstyle:hiddenfield")
+ public Builder setRuntimeName(final String runtimeName) {
+ this.runtimeName = runtimeName;
+ return this;
+ }
+
+ /**
* Set number of cores.
*
* @param cores the number of cores
@@ -215,7 +249,7 @@
*/
@Override
public EvaluatorRequest build() {
- return new EvaluatorRequest(this.n, this.megaBytes, this.cores, this.nodeNames, this.rackNames);
+ return new EvaluatorRequest(this.n, this.megaBytes, this.cores, this.nodeNames, this.rackNames, this.runtimeName);
}
}
}
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/EvaluatorRequestorImpl.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/EvaluatorRequestorImpl.java
index e433513..b5dc69c 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/EvaluatorRequestorImpl.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/EvaluatorRequestorImpl.java
@@ -77,7 +77,9 @@
if (req.getRackNames() == null) {
throw new IllegalArgumentException("Rack names cannot be null");
}
-
+ if(req.getRuntimeName() == null) {
+ throw new IllegalArgumentException("Runtime name cannot be null");
+ }
// for backwards compatibility, we will always set the relax locality flag
// to true unless the user configured racks, in which case we will check for
// the ANY modifier (*), if not there, then we won't relax the locality
@@ -105,6 +107,7 @@
.addNodeNames(req.getNodeNames())
.addRackNames(req.getRackNames())
.setRelaxLocality(relaxLocality)
+ .setRuntimeName(req.getRuntimeName())
.build();
this.resourceRequestHandler.onNext(request);
}
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEvent.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEvent.java
index 592d01c..f2f2c51 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEvent.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEvent.java
@@ -60,4 +60,9 @@
* @return List of libraries local to this Evaluator
*/
Set<FileResource> getFileSet();
+
+ /**
+ * @return name of the runtime to launch the Evaluator on
+ */
+ String getRuntimeName();
}
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEventImpl.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEventImpl.java
index 1b9e943..bacc563 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEventImpl.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceLaunchEventImpl.java
@@ -40,6 +40,7 @@
private final Configuration evaluatorConf;
private final EvaluatorProcess process;
private final Set<FileResource> fileSet;
+ private final String runtimeName;
private ResourceLaunchEventImpl(final Builder builder) {
this.identifier = BuilderUtils.notNull(builder.identifier);
@@ -47,6 +48,7 @@
this.evaluatorConf = BuilderUtils.notNull(builder.evaluatorConf);
this.process = BuilderUtils.notNull(builder.process);
this.fileSet = BuilderUtils.notNull(builder.fileSet);
+ this.runtimeName = BuilderUtils.notNull(builder.runtimeName);
}
@Override
@@ -74,6 +76,11 @@
return fileSet;
}
+ @Override
+ public String getRuntimeName() {
+ return runtimeName;
+ }
+
public static Builder newBuilder() {
return new Builder();
}
@@ -87,6 +94,7 @@
private Configuration evaluatorConf;
private EvaluatorProcess process;
private Set<FileResource> fileSet = new HashSet<>();
+ private String runtimeName;
/**
* @see ResourceLaunchEvent#getIdentifier()
@@ -97,6 +105,14 @@
}
/**
+ * @see ResourceLaunchEvent#getRuntimeName()
+ */
+ public Builder setRuntimeName(final String runtimeName) {
+ this.runtimeName = runtimeName;
+ return this;
+ }
+
+ /**
* @see ResourceLaunchEvent#getRemoteId()
*/
public Builder setRemoteId(final String remoteId) {
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEvent.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEvent.java
index 30d9464..c3275ce 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEvent.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEvent.java
@@ -34,4 +34,9 @@
* @return Id of the resource to release
*/
String getIdentifier();
+
+ /**
+ * @return name of the runtime that this resource belongs to
+ */
+ String getRuntimeName();
}
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEventImpl.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEventImpl.java
index 4778f84..9aaa0dd 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEventImpl.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceReleaseEventImpl.java
@@ -27,9 +27,11 @@
public final class ResourceReleaseEventImpl implements ResourceReleaseEvent {
private final String identifier;
+ private final String runtimeName;
private ResourceReleaseEventImpl(final Builder builder) {
this.identifier = BuilderUtils.notNull(builder.identifier);
+ this.runtimeName = BuilderUtils.notNull(builder.runtimeName);
}
@Override
@@ -37,6 +39,11 @@
return identifier;
}
+ @Override
+ public String getRuntimeName() {
+ return runtimeName;
+ }
+
public static Builder newBuilder() {
return new Builder();
}
@@ -47,7 +54,7 @@
public static final class Builder implements org.apache.reef.util.Builder<ResourceReleaseEvent> {
private String identifier;
-
+ private String runtimeName;
/**
* @see ResourceReleaseEvent#getIdentifier()
*/
@@ -56,6 +63,14 @@
return this;
}
+ /**
+ * @see ResourceReleaseEvent#getRuntimeName()
+ */
+ public Builder setRuntimeName(final String runtimeName) {
+ this.runtimeName = runtimeName;
+ return this;
+ }
+
@Override
public ResourceReleaseEvent build() {
return new ResourceReleaseEventImpl(this);
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEvent.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEvent.java
index 8cf8a33..b8ce952 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEvent.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEvent.java
@@ -71,4 +71,9 @@
* the preferred list. If false, strictly enforce the preferences.
*/
Optional<Boolean> getRelaxLocality();
+
+ /**
+ * @return The runtime name
+ */
+ String getRuntimeName();
}
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEventImpl.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEventImpl.java
index 38ded64..f385087 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEventImpl.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/api/ResourceRequestEventImpl.java
@@ -36,6 +36,7 @@
private final Optional<Integer> priority;
private final Optional<Integer> virtualCores;
private final Optional<Boolean> relaxLocality;
+ private final String runtimeName;
private ResourceRequestEventImpl(final Builder builder) {
this.resourceCount = BuilderUtils.notNull(builder.resourceCount);
@@ -45,6 +46,7 @@
this.priority = Optional.ofNullable(builder.priority);
this.virtualCores = Optional.ofNullable(builder.virtualCores);
this.relaxLocality = Optional.ofNullable(builder.relaxLocality);
+ this.runtimeName = builder.runtimeName == null ? "" : builder.runtimeName;
}
@Override
@@ -82,6 +84,11 @@
return relaxLocality;
}
+ @Override
+ public String getRuntimeName() {
+ return runtimeName;
+ }
+
public static Builder newBuilder() {
return new Builder();
}
@@ -97,6 +104,7 @@
private Integer priority;
private Integer virtualCores;
private Boolean relaxLocality;
+ private String runtimeName;
/**
* Create a builder from an existing ResourceRequestEvent.
@@ -109,6 +117,7 @@
this.priority = resourceRequestEvent.getPriority().orElse(null);
this.virtualCores = resourceRequestEvent.getVirtualCores().orElse(null);
this.relaxLocality = resourceRequestEvent.getRelaxLocality().orElse(null);
+ this.runtimeName = resourceRequestEvent.getRuntimeName();
return this;
}
@@ -192,6 +201,14 @@
return this;
}
+ /**
+ * @see ResourceRequestEvent#getRuntimeName
+ */
+ public Builder setRuntimeName(final String runtimeName) {
+ this.runtimeName = runtimeName;
+ return this;
+ }
+
@Override
public ResourceRequestEvent build() {
return new ResourceRequestEventImpl(this);
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/AllocatedEvaluatorImpl.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/AllocatedEvaluatorImpl.java
index c6d9b5d..e57d7a8 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/AllocatedEvaluatorImpl.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/AllocatedEvaluatorImpl.java
@@ -244,7 +244,8 @@
.setRemoteId(this.remoteID)
.setEvaluatorConf(evaluatorConfiguration)
.addFiles(this.files)
- .addLibraries(this.libraries);
+ .addLibraries(this.libraries)
+ .setRuntimeName(this.getEvaluatorDescriptor().getRuntimeName());
rbuilder.setProcess(this.evaluatorManager.getEvaluatorDescriptor().getProcess());
this.evaluatorManager.onResourceLaunch(rbuilder.build());
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
index fcefb6b..9003ca4 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
@@ -244,16 +244,20 @@
@Override
public void onNext(final Alarm alarm) {
EvaluatorManager.this.resourceReleaseHandler.onNext(
- ResourceReleaseEventImpl.newBuilder()
- .setIdentifier(EvaluatorManager.this.evaluatorId).build()
+ ResourceReleaseEventImpl.newBuilder()
+ .setIdentifier(EvaluatorManager.this.evaluatorId)
+ .setRuntimeName(EvaluatorManager.this.getEvaluatorDescriptor().getRuntimeName())
+ .build()
);
}
});
} catch (final IllegalStateException e) {
LOG.log(Level.WARNING, "Force resource release because the client closed the clock.", e);
EvaluatorManager.this.resourceReleaseHandler.onNext(
- ResourceReleaseEventImpl.newBuilder()
- .setIdentifier(EvaluatorManager.this.evaluatorId).build()
+ ResourceReleaseEventImpl.newBuilder()
+ .setIdentifier(EvaluatorManager.this.evaluatorId)
+ .setRuntimeName(EvaluatorManager.this.getEvaluatorDescriptor().getRuntimeName())
+ .build()
);
}
}
