You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by yi...@apache.org on 2014/11/11 17:59:23 UTC
incubator-reef git commit: This address [REEF-6] and [REEF-25]. It
changes the policy of sending requests to YARN RM from
"one-request-at-a-time" to "one-kind-of-requests-at-a-time". Contributed by
John Yang Closes #4
Repository: incubator-reef
Updated Branches:
refs/heads/master 10f965143 -> 16dd52a35
This address [REEF-6] and [REEF-25].
It changes the policy of sending requests to YARN RM from "one-request-at-a-time" to "one-kind-of-requests-at-a-time".
Contributed by John Yang <jo...@apache.org>
Closes #4
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/16dd52a3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/16dd52a3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/16dd52a3
Branch: refs/heads/master
Commit: 16dd52a353f3ac57695e50670440c85b5f385829
Parents: 10f9651
Author: John Yang <jo...@gmail.com>
Authored: Fri Oct 24 15:36:24 2014 +0900
Committer: Yingda Chen <yd...@gmail.com>
Committed: Mon Nov 10 10:41:42 2014 -0800
----------------------------------------------------------------------
.../javabridge/EvaluatorRequestorBridge.java | 35 +------
.../yarn/driver/YarnContainerManager.java | 105 ++++++++++++-------
2 files changed, 68 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/16dd52a3/reef-bridge-project/reef-bridge-java/src/main/java/org/apache/reef/javabridge/EvaluatorRequestorBridge.java
----------------------------------------------------------------------
diff --git a/reef-bridge-project/reef-bridge-java/src/main/java/org/apache/reef/javabridge/EvaluatorRequestorBridge.java b/reef-bridge-project/reef-bridge-java/src/main/java/org/apache/reef/javabridge/EvaluatorRequestorBridge.java
index 18b3d1e..cfabfde 100644
--- a/reef-bridge-project/reef-bridge-java/src/main/java/org/apache/reef/javabridge/EvaluatorRequestorBridge.java
+++ b/reef-bridge-project/reef-bridge-java/src/main/java/org/apache/reef/javabridge/EvaluatorRequestorBridge.java
@@ -18,21 +18,14 @@
*/
package org.apache.reef.javabridge;
-import org.apache.reef.driver.catalog.RackDescriptor;
import org.apache.reef.driver.evaluator.EvaluatorRequest;
import org.apache.reef.driver.evaluator.EvaluatorRequestor;
-import org.apache.reef.proto.DriverRuntimeProtocol;
-import org.apache.reef.runtime.common.driver.catalog.ResourceCatalogImpl;
-import org.apache.reef.tang.Tang;
-import org.apache.reef.tang.exceptions.InjectionException;
import java.util.logging.Level;
import java.util.logging.Logger;
public final class EvaluatorRequestorBridge extends NativeBridge {
private static final Logger LOG = Logger.getLogger(EvaluatorRequestorBridge.class.getName());
- private static final String REQUEST_CONTAINERS_ON_RACK = "*"; // The rack used for all requests
- private static final String REQUEST_CONTAINER_ON_HOST = "HostName"; // The hostname used for all requests.
private final boolean isBlocked;
private final EvaluatorRequestor jevaluatorRequestor;
// accumulate how many evaluators have been submitted through this instance
@@ -51,22 +44,12 @@ public final class EvaluatorRequestorBridge extends NativeBridge {
}
if (rack != null && !rack.isEmpty()) {
- LOG.log(Level.WARNING, "Ignoring rack request and using [{0}] instead.", REQUEST_CONTAINERS_ON_RACK);
+ LOG.log(Level.WARNING, "Ignoring rack preference.");
}
clrEvaluatorsNumber += evaluatorsNumber;
- final ResourceCatalogImpl catalog = getNewResourceCatalogInstance();
-
- catalog.handle(DriverRuntimeProtocol.NodeDescriptorProto.newBuilder()
- .setRackName(REQUEST_CONTAINERS_ON_RACK)
- .setHostName(REQUEST_CONTAINER_ON_HOST)
- .setPort(0)
- .setMemorySize(memory)
- .setIdentifier("clrBridgeRackCatalog")
- .build());
- final RackDescriptor rackDescriptor = (RackDescriptor) (catalog.getRacks().toArray())[0];
- final EvaluatorRequest request = EvaluatorRequest.newBuilder().fromDescriptor(rackDescriptor)
+ final EvaluatorRequest request = EvaluatorRequest.newBuilder()
.setNumber(evaluatorsNumber)
.setMemory(memory)
.setNumberOfCores(virtualCore)
@@ -80,20 +63,6 @@ public final class EvaluatorRequestorBridge extends NativeBridge {
return clrEvaluatorsNumber;
}
- /**
- * Helper method that uses Tang to generate a new instance of ResourceCatalogImpl
- *
- * @return a new instance of ResourceCatalogImpl
- */
- private ResourceCatalogImpl getNewResourceCatalogInstance() {
- try {
- return Tang.Factory.getTang().newInjector().getInstance(ResourceCatalogImpl.class);
- } catch (final InjectionException e) {
- LOG.log(Level.SEVERE, "Cannot inject resource catalog", e);
- throw new RuntimeException("Cannot inject resource catalog");
- }
- }
-
@Override
public void close() {
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/16dd52a3/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
----------------------------------------------------------------------
diff --git a/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java b/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
index fffd942..3f922dc 100644
--- a/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
+++ b/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
@@ -19,6 +19,7 @@
package org.apache.reef.runtime.yarn.driver;
import com.google.protobuf.ByteString;
+import org.apache.commons.collections.ListUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -51,6 +52,7 @@ import javax.inject.Inject;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -68,8 +70,11 @@ final class YarnContainerManager
private final YarnClient yarnClient = YarnClient.createYarnClient();
- private final Queue<AMRMClient.ContainerRequest>
- outstandingContainerRequests = new ConcurrentLinkedQueue<>();
+ private final Queue<AMRMClient.ContainerRequest> requestsBeforeSentToRM = new ConcurrentLinkedQueue<>();
+
+ private final Queue<AMRMClient.ContainerRequest> requestsAfterSentToRM = new ConcurrentLinkedQueue<>();
+
+ private final Map<String, String> nodeIdToRackName = new ConcurrentHashMap<>();
private final YarnConfiguration yarnConf;
private final AMRMClientAsync resourceManager;
@@ -81,8 +86,6 @@ final class YarnContainerManager
private final DriverStatusManager driverStatusManager;
private final TrackingURLProvider trackingURLProvider;
- private int lastContainerMemory = -1;
-
@Inject
YarnContainerManager(
final YarnConfiguration yarnConf,
@@ -145,6 +148,7 @@ final class YarnContainerManager
@Override
public void onNodesUpdated(final List<NodeReport> nodeReports) {
for (final NodeReport nodeReport : nodeReports) {
+ this.nodeIdToRackName.put(nodeReport.getNodeId().toString(), nodeReport.getRackName());
onNodeReport(nodeReport);
}
}
@@ -399,18 +403,8 @@ final class YarnContainerManager
synchronized (this) {
this.containerRequestCounter.incrementBy(containerRequests.length);
- boolean queueWasEmpty = this.outstandingContainerRequests.isEmpty();
- for (final AMRMClient.ContainerRequest containerRequest : containerRequests) {
- LOG.log(Level.FINEST, "Container Request: memory = {0}, core number = {1}", new Object[]{containerRequest.getCapability().getMemory(), containerRequest.getCapability().getVirtualCores()});
- LOG.log(Level.FINEST, "Adding container request to local queue: {0}", containerRequest);
- this.outstandingContainerRequests.add(containerRequest);
- }
- if (queueWasEmpty && containerRequests.length != 0) {
- AMRMClient.ContainerRequest firstRequest = outstandingContainerRequests.peek();
- LOG.log(Level.INFO, "Requesting first container from YARN: {0}", firstRequest);
- this.lastContainerMemory = firstRequest.getCapability().getMemory();
- this.resourceManager.addContainerRequest(firstRequest);
- }
+ this.requestsBeforeSentToRM.addAll(Arrays.asList(containerRequests));
+ doHomogeneousRequests();
}
this.updateRuntimeStatus();
@@ -427,31 +421,29 @@ final class YarnContainerManager
// recovered container is not new allocation, it is just checking back from previous driver failover
if (!isRecoveredContainer) {
synchronized (this) {
- this.containerRequestCounter.decrement();
- if (!this.outstandingContainerRequests.isEmpty()) {
+ if (matchContainerWithPendingRequest(container)) {
+ final AMRMClient.ContainerRequest matchedRequest = this.requestsAfterSentToRM.peek();
+ this.containerRequestCounter.decrement();
this.containers.add(container);
- final AMRMClient.ContainerRequest requestToBeSubmitted = this.outstandingContainerRequests.peek();
- final int currentRequestMemory = requestToBeSubmitted.getCapability().getMemory();
- AMRMClient.ContainerRequest removedRequest = this.outstandingContainerRequests.remove();
- if (this.lastContainerMemory == currentRequestMemory) {
- // to be safe we need to make sure that previous request is no longer in async AMRM client's queue
- // it is ok if the request is no longer there
+ LOG.log(Level.FINEST, "{0} matched with {1}", new Object[]{container.toString(), matchedRequest.toString()});
+
+ // Due to the bug YARN-314 and the workings of AMRMCClientAsync, when x-priority m-capacity zero-container request
+ // and x-priority n-capacity nonzero-container request are sent together, where m > n, RM ignores the latter.
+ // Therefore it is necessary avoid sending zero-container request, even it means getting extra containers.
+ // It is okay to send nonzero m-capacity and n-capacity request together since bigger containers can be matched.
+ // TODO: revisit this when implementing locality-strictness (i.e. a specific rack request can be ignored)
+ if (this.requestsAfterSentToRM.size() > 1) {
try {
- LOG.log(Level.INFO,
- "Requesting same amount of memory {0}, make sure previous request {1} is removed from AMRM client queue",
- new Object[]{currentRequestMemory, removedRequest});
- this.resourceManager.removeContainerRequest(removedRequest);
+ this.resourceManager.removeContainerRequest(matchedRequest);
} catch (final Exception e) {
LOG.log(Level.WARNING, "Nothing to remove from Async AMRM client's queue, removal attempt failed with exception", e);
}
}
- if (requestToBeSubmitted != null) {
- LOG.log(Level.INFO, "Requesting 1 additional container from YARN: {0}", requestToBeSubmitted);
- this.resourceManager.addContainerRequest(requestToBeSubmitted);
- this.lastContainerMemory = currentRequestMemory;
- }
+ this.requestsAfterSentToRM.remove();
+ doHomogeneousRequests();
+
LOG.log(Level.FINEST, "Allocated Container: memory = {0}, core number = {1}", new Object[]{container.getResource().getMemory(), container.getResource().getVirtualCores()});
this.reefEventHandlers.onResourceAllocation(ResourceAllocationProto.newBuilder()
.setIdentifier(container.getId().toString())
@@ -463,18 +455,53 @@ final class YarnContainerManager
logContainerAddition(container.getId().toString());
this.updateRuntimeStatus();
} else {
- // since no request is removed from local queue until new container is allocated
- // the queue should not be empty at the beginning of this call, as a failsafe, we release the
- // unexpected container
- LOG.log(Level.WARNING,
- "outstandingContainerRequests is empty upon allocation of unexpected container {0}, releasing...",
- container.getId());
+ LOG.log(Level.WARNING, "Got an extra container {0} that doesn't match, releasing...", container.getId());
this.resourceManager.releaseAssignedContainer(container.getId());
}
}
}
}
+ private synchronized void doHomogeneousRequests() {
+ if (this.requestsAfterSentToRM.isEmpty()) {
+ final AMRMClient.ContainerRequest firstRequest = this.requestsBeforeSentToRM.peek();
+
+ while (!this.requestsBeforeSentToRM.isEmpty() && isSameKindOfRequest(firstRequest, this.requestsBeforeSentToRM.peek())) {
+ final AMRMClient.ContainerRequest homogeneousRequest = this.requestsBeforeSentToRM.remove();
+ this.resourceManager.addContainerRequest(homogeneousRequest);
+ this.requestsAfterSentToRM.add(homogeneousRequest);
+ }
+ }
+ }
+
+ private boolean isSameKindOfRequest(AMRMClient.ContainerRequest r1, AMRMClient.ContainerRequest r2) {
+ return r1.getPriority().compareTo(r2.getPriority()) == 0
+ && r1.getCapability().compareTo(r2.getCapability()) == 0
+ && r1.getRelaxLocality() == r2.getRelaxLocality()
+ && ListUtils.isEqualList(r1.getNodes(), r2.getNodes())
+ && ListUtils.isEqualList(r1.getRacks(), r2.getRacks());
+ }
+
+ /**
+ * Match to see whether the container satisfies the request.
+ * We take into consideration that RM has some freedom in rounding
+ * up the allocation and in placing containers on other machines.
+ */
+ private boolean matchContainerWithPendingRequest(Container container) {
+ if (this.requestsAfterSentToRM.isEmpty()) {
+ return false;
+ }
+
+ final AMRMClient.ContainerRequest request = this.requestsAfterSentToRM.peek();
+ final boolean resourceCondition = container.getResource().getMemory() >= request.getCapability().getMemory(); // TODO: check vcores once YARN-2380 is resolved
+ final boolean nodeCondition = request.getNodes() == null
+ || request.getNodes().contains(container.getNodeId().getHost());
+ final boolean rackCondition = request.getRacks() == null
+ || request.getRacks().contains(this.nodeIdToRackName.get(container.getNodeId().toString()));
+
+ return resourceCondition && (request.getRelaxLocality() || (rackCondition && nodeCondition));
+ }
+
/**
* Update the driver with my current status
*/