添加Capacity调度器的源码注释
This commit is contained in:
parent
fbd6e04709
commit
dce4387b3d
@ -22,7 +22,12 @@
|
|||||||
* Resource classification.
|
* Resource classification.
|
||||||
*/
|
*/
|
||||||
public enum NodeType {
|
public enum NodeType {
|
||||||
NODE_LOCAL(0), RACK_LOCAL(1), OFF_SWITCH(2);
|
//请求规定了必须运行在某个的服务器节点
|
||||||
|
NODE_LOCAL(0),
|
||||||
|
// 请求规定了必须运行在某个机架上
|
||||||
|
RACK_LOCAL(1),
|
||||||
|
// 这个请求对本地化没有要求
|
||||||
|
OFF_SWITCH(2);
|
||||||
|
|
||||||
private final int index;
|
private final int index;
|
||||||
|
|
||||||
|
@ -1363,6 +1363,7 @@ protected void nodeUpdate(RMNode rmNode) {
|
|||||||
readLock.lock();
|
readLock.lock();
|
||||||
try {
|
try {
|
||||||
setLastNodeUpdateTime(Time.now());
|
setLastNodeUpdateTime(Time.now());
|
||||||
|
// 优先处理心跳逻辑。
|
||||||
super.nodeUpdate(rmNode);
|
super.nodeUpdate(rmNode);
|
||||||
} finally {
|
} finally {
|
||||||
readLock.unlock();
|
readLock.unlock();
|
||||||
@ -1370,6 +1371,7 @@ protected void nodeUpdate(RMNode rmNode) {
|
|||||||
|
|
||||||
// Try to do scheduling
|
// Try to do scheduling
|
||||||
if (!scheduleAsynchronously) {
|
if (!scheduleAsynchronously) {
|
||||||
|
// 在不开启异步调度的时候,需要借助节点的心跳进行调度。默认建议开启异步调度。
|
||||||
writeLock.lock();
|
writeLock.lock();
|
||||||
try {
|
try {
|
||||||
// reset allocation and reservation stats before we start doing any
|
// reset allocation and reservation stats before we start doing any
|
||||||
@ -1484,6 +1486,13 @@ private void updateSchedulerHealth(long now, NodeId nodeId,
|
|||||||
.getAssignmentInformation().getReserved());
|
.getAssignmentInformation().getReserved());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 是否需要继续分配资源。
|
||||||
|
* @param assignment 已经分配的资源信息
|
||||||
|
* @param offswitchCount 不限制节点和机架的个数。
|
||||||
|
* @param assignedContainers 已经分配的Container个数
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
private boolean canAllocateMore(CSAssignment assignment, int offswitchCount,
|
private boolean canAllocateMore(CSAssignment assignment, int offswitchCount,
|
||||||
int assignedContainers) {
|
int assignedContainers) {
|
||||||
// Current assignment shouldn't be empty
|
// Current assignment shouldn't be empty
|
||||||
@ -1529,6 +1538,7 @@ private CandidateNodeSet<FiCaSchedulerNode> getCandidateNodeSet(
|
|||||||
/**
|
/**
|
||||||
* We need to make sure when doing allocation, Node should be existed
|
* We need to make sure when doing allocation, Node should be existed
|
||||||
* And we will construct a {@link CandidateNodeSet} before proceeding
|
* And we will construct a {@link CandidateNodeSet} before proceeding
|
||||||
|
* @param withNodeHeartbeat 如果是未开启异步调度,则需要通过心跳来实现,当前值为true,否则为false
|
||||||
*/
|
*/
|
||||||
private void allocateContainersToNode(NodeId nodeId,
|
private void allocateContainersToNode(NodeId nodeId,
|
||||||
boolean withNodeHeartbeat) {
|
boolean withNodeHeartbeat) {
|
||||||
@ -1544,6 +1554,7 @@ private void allocateContainersToNode(NodeId nodeId,
|
|||||||
// Only check if we can allocate more container on the same node when
|
// Only check if we can allocate more container on the same node when
|
||||||
// scheduling is triggered by node heartbeat
|
// scheduling is triggered by node heartbeat
|
||||||
if (null != assignment && withNodeHeartbeat) {
|
if (null != assignment && withNodeHeartbeat) {
|
||||||
|
// 如果是由心跳触发的分配,会尝试再次分配。
|
||||||
if (assignment.getType() == NodeType.OFF_SWITCH) {
|
if (assignment.getType() == NodeType.OFF_SWITCH) {
|
||||||
offswitchCount++;
|
offswitchCount++;
|
||||||
}
|
}
|
||||||
@ -1693,6 +1704,12 @@ private void allocateFromReservedContainer(FiCaSchedulerNode node,
|
|||||||
submitResourceCommitRequest(getClusterResource(), assignment);
|
submitResourceCommitRequest(getClusterResource(), assignment);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 为Container申请或者预留资源。
|
||||||
|
* @param candidates
|
||||||
|
* @param withNodeHeartbeat
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
private CSAssignment allocateOrReserveNewContainers(
|
private CSAssignment allocateOrReserveNewContainers(
|
||||||
CandidateNodeSet<FiCaSchedulerNode> candidates,
|
CandidateNodeSet<FiCaSchedulerNode> candidates,
|
||||||
boolean withNodeHeartbeat) {
|
boolean withNodeHeartbeat) {
|
||||||
@ -1787,6 +1804,12 @@ private CSAssignment allocateContainersOnMultiNodes(
|
|||||||
return allocateOrReserveNewContainers(candidates, false);
|
return allocateOrReserveNewContainers(candidates, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 将资源分派给指定的节点
|
||||||
|
* @param candidates
|
||||||
|
* @param withNodeHeartbeat
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
@VisibleForTesting
|
@VisibleForTesting
|
||||||
CSAssignment allocateContainersToNode(
|
CSAssignment allocateContainersToNode(
|
||||||
CandidateNodeSet<FiCaSchedulerNode> candidates,
|
CandidateNodeSet<FiCaSchedulerNode> candidates,
|
||||||
|
Loading…
Reference in New Issue
Block a user