DefaultPlacementAlgorithm.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.algorithm;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.InvalidAllocationTagsQueryException;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.PlacementConstraintManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.PlacementConstraintsUtil;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.ConstraintPlacementAlgorithm;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.ConstraintPlacementAlgorithmInput;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.ConstraintPlacementAlgorithmOutput;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.ConstraintPlacementAlgorithmOutputCollector;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.PlacedSchedulingRequest;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.api.SchedulingRequestWithPlacementAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.processor.BatchedRequests;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.constraint.processor.NodeCandidateSelector;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Basic placement algorithm.
* Supports different Iterators at SchedulingRequest level including:
* Serial, PopularTags
*/
public class DefaultPlacementAlgorithm implements ConstraintPlacementAlgorithm {
private static final Logger LOG =
LoggerFactory.getLogger(DefaultPlacementAlgorithm.class);
// Number of times to re-attempt placing a single scheduling request.
private static final int RE_ATTEMPT_COUNT = 2;
private LocalAllocationTagsManager tagsManager;
private PlacementConstraintManager constraintManager;
private NodeCandidateSelector nodeSelector;
private ResourceCalculator resourceCalculator;
@Override
public void init(RMContext rmContext) {
this.tagsManager = new LocalAllocationTagsManager(
rmContext.getAllocationTagsManager());
this.constraintManager = rmContext.getPlacementConstraintManager();
this.resourceCalculator = rmContext.getScheduler().getResourceCalculator();
this.nodeSelector =
filter -> ((AbstractYarnScheduler) (rmContext).getScheduler())
.getNodes(filter);
}
boolean attemptPlacementOnNode(ApplicationId appId,
Resource availableResources, SchedulingRequest schedulingRequest,
SchedulerNode schedulerNode, boolean ignoreResourceCheck)
throws InvalidAllocationTagsQueryException {
boolean fitsInNode = ignoreResourceCheck ||
Resources.fitsIn(resourceCalculator,
schedulingRequest.getResourceSizing().getResources(),
availableResources);
boolean constraintsSatisfied =
PlacementConstraintsUtil.canSatisfyConstraints(appId,
schedulingRequest, schedulerNode, constraintManager, tagsManager);
return fitsInNode && constraintsSatisfied;
}
@Override
public void place(ConstraintPlacementAlgorithmInput input,
ConstraintPlacementAlgorithmOutputCollector collector) {
BatchedRequests requests = (BatchedRequests) input;
int placementAttempt = requests.getPlacementAttempt();
ConstraintPlacementAlgorithmOutput resp =
new ConstraintPlacementAlgorithmOutput(requests.getApplicationId());
List<SchedulerNode> allNodes = nodeSelector.selectNodes(null);
List<SchedulingRequest> rejectedRequests = new ArrayList<>();
Map<NodeId, Resource> availResources = new HashMap<>();
int rePlacementCount = RE_ATTEMPT_COUNT;
while (rePlacementCount > 0) {
doPlacement(requests, resp, allNodes, rejectedRequests, availResources);
// Double check if placement constraints are really satisfied
validatePlacement(requests.getApplicationId(), resp,
rejectedRequests, availResources);
if (rejectedRequests.size() == 0 || rePlacementCount == 1) {
break;
}
requests = new BatchedRequests(requests.getIteratorType(),
requests.getApplicationId(), rejectedRequests,
requests.getPlacementAttempt());
rejectedRequests = new ArrayList<>();
rePlacementCount--;
}
resp.getRejectedRequests().addAll(
rejectedRequests.stream().map(
x -> new SchedulingRequestWithPlacementAttempt(
placementAttempt, x)).collect(Collectors.toList()));
collector.collect(resp);
// Clean current temp-container tags
this.tagsManager.cleanTempContainers(requests.getApplicationId());
}
private void doPlacement(BatchedRequests requests,
ConstraintPlacementAlgorithmOutput resp,
List<SchedulerNode> allNodes,
List<SchedulingRequest> rejectedRequests,
Map<NodeId, Resource> availableResources) {
Iterator<SchedulingRequest> requestIterator = requests.iterator();
Iterator<SchedulerNode> nIter = allNodes.iterator();
SchedulerNode lastSatisfiedNode = null;
while (requestIterator.hasNext()) {
if (allNodes.isEmpty()) {
LOG.warn("No nodes available for placement at the moment !!");
break;
}
SchedulingRequest schedulingRequest = requestIterator.next();
PlacedSchedulingRequest placedReq =
new PlacedSchedulingRequest(schedulingRequest);
placedReq.setPlacementAttempt(requests.getPlacementAttempt());
resp.getPlacedRequests().add(placedReq);
CircularIterator<SchedulerNode> nodeIter =
new CircularIterator(lastSatisfiedNode, nIter, allNodes);
int numAllocs =
schedulingRequest.getResourceSizing().getNumAllocations();
while (nodeIter.hasNext() && numAllocs > 0) {
SchedulerNode node = nodeIter.next();
try {
String tag = schedulingRequest.getAllocationTags() == null ? "" :
schedulingRequest.getAllocationTags().iterator().next();
Resource unallocatedResource =
availableResources.computeIfAbsent(node.getNodeID(),
x -> Resource.newInstance(node.getUnallocatedResource()));
if (!requests.getBlacklist(tag).contains(node.getNodeID()) &&
attemptPlacementOnNode(
requests.getApplicationId(), unallocatedResource,
schedulingRequest, node, false)) {
schedulingRequest.getResourceSizing()
.setNumAllocations(--numAllocs);
Resources.addTo(unallocatedResource,
schedulingRequest.getResourceSizing().getResources());
placedReq.getNodes().add(node);
numAllocs =
schedulingRequest.getResourceSizing().getNumAllocations();
// Add temp-container tags for current placement cycle
this.tagsManager.addTempTags(node.getNodeID(),
requests.getApplicationId(),
schedulingRequest.getAllocationTags());
lastSatisfiedNode = node;
}
} catch (InvalidAllocationTagsQueryException e) {
LOG.warn("Got exception from TagManager !", e);
}
}
}
// Add all requests whose numAllocations still > 0 to rejected list.
requests.getSchedulingRequests().stream()
.filter(sReq -> sReq.getResourceSizing().getNumAllocations() > 0)
.forEach(rejReq -> rejectedRequests.add(cloneReq(rejReq)));
}
/**
* During the placement phase, allocation tags are added to the node if the
* constraint is satisfied, But depending on the order in which the
* algorithm sees the request, it is possible that a constraint that happened
* to be valid during placement of an earlier-seen request, might not be
* valid after all subsequent requests have been placed.
*
* For eg:
* Assume nodes n1, n2, n3, n4 and n5
*
* Consider the 2 constraints:
* 1) "foo", anti-affinity with "foo"
* 2) "bar", anti-affinity with "foo"
*
* And 2 requests
* req1: NumAllocations = 4, allocTags = [foo]
* req2: NumAllocations = 1, allocTags = [bar]
*
* If "req1" is seen first, the algorithm can place the 4 containers in
* n1, n2, n3 and n4. And when it gets to "req2", it will see that 4 nodes
* with the "foo" tag and will place on n5.
* But if "req2" is seem first, then "bar" will be placed on any node,
* since no node currently has "foo", and when it gets to "req1", since
* "foo" has not anti-affinity with "bar", the algorithm can end up placing
* "foo" on a node with "bar" violating the second constraint.
*
* To prevent the above, we need a validation step: after the placements for a
* batch of requests are made, for each req, we remove its tags from the node
* and try to see of constraints are still satisfied if the tag were to be
* added back on the node.
*
* When applied to the example above, after "req2" and "req1" are placed,
* we remove the "bar" tag from the node and try to add it back on the node.
* This time, constraint satisfaction will fail, since there is now a "foo"
* tag on the node and "bar" cannot be added. The algorithm will then
* retry placing "req2" on another node.
*
* @param applicationId
* @param resp
* @param rejectedRequests
* @param availableResources
*/
private void validatePlacement(ApplicationId applicationId,
ConstraintPlacementAlgorithmOutput resp,
List<SchedulingRequest> rejectedRequests,
Map<NodeId, Resource> availableResources) {
Iterator<PlacedSchedulingRequest> pReqIter =
resp.getPlacedRequests().iterator();
while (pReqIter.hasNext()) {
PlacedSchedulingRequest pReq = pReqIter.next();
Iterator<SchedulerNode> nodeIter = pReq.getNodes().iterator();
// Assuming all reqs were satisfied.
int num = 0;
while (nodeIter.hasNext()) {
SchedulerNode node = nodeIter.next();
try {
// Remove just the tags for this placement.
this.tagsManager.removeTempTags(node.getNodeID(),
applicationId, pReq.getSchedulingRequest().getAllocationTags());
Resource availOnNode = availableResources.get(node.getNodeID());
if (!attemptPlacementOnNode(applicationId, availOnNode,
pReq.getSchedulingRequest(), node, true)) {
nodeIter.remove();
num++;
Resources.subtractFrom(availOnNode,
pReq.getSchedulingRequest().getResourceSizing().getResources());
} else {
// Add back the tags if everything is fine.
this.tagsManager.addTempTags(node.getNodeID(),
applicationId, pReq.getSchedulingRequest().getAllocationTags());
}
} catch (InvalidAllocationTagsQueryException e) {
LOG.warn("Got exception from TagManager !", e);
}
}
if (num > 0) {
SchedulingRequest sReq = cloneReq(pReq.getSchedulingRequest());
sReq.getResourceSizing().setNumAllocations(num);
rejectedRequests.add(sReq);
}
if (pReq.getNodes().isEmpty()) {
pReqIter.remove();
}
}
}
private static SchedulingRequest cloneReq(SchedulingRequest sReq) {
return SchedulingRequest.newInstance(
sReq.getAllocationRequestId(), sReq.getPriority(),
sReq.getExecutionType(), sReq.getAllocationTags(),
ResourceSizing.newInstance(
sReq.getResourceSizing().getNumAllocations(),
sReq.getResourceSizing().getResources()),
sReq.getPlacementConstraint());
}
}