YARN-11015. Decouple queue capacity with ability to run OPPORTUNISTIC container #3779
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,6 +24,7 @@ | |
| import org.apache.hadoop.yarn.api.records.ContainerExitStatus; | ||
| import org.apache.hadoop.yarn.api.records.ContainerId; | ||
| import org.apache.hadoop.yarn.api.records.ExecutionType; | ||
| import org.apache.hadoop.yarn.api.records.Resource; | ||
| import org.apache.hadoop.yarn.api.records.ResourceUtilization; | ||
| import org.apache.hadoop.yarn.conf.YarnConfiguration; | ||
| import org.apache.hadoop.yarn.event.AsyncDispatcher; | ||
|
|
@@ -46,6 +47,7 @@ | |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService | ||
| .RecoveredContainerState; | ||
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerStatus; | ||
| import org.apache.hadoop.yarn.util.resource.Resources; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
|
|
@@ -74,6 +76,7 @@ public class ContainerScheduler extends AbstractService implements | |
| private final Context context; | ||
| // Capacity of the queue for opportunistic Containers. | ||
| private final int maxOppQueueLength; | ||
| private final boolean forceStartGuaranteedContainers; | ||
|
|
||
| // Queue of Guaranteed Containers waiting for resources to run | ||
| private final LinkedHashMap<ContainerId, Container> | ||
|
|
@@ -106,9 +109,39 @@ public class ContainerScheduler extends AbstractService implements | |
|
|
||
| private final AsyncDispatcher dispatcher; | ||
| private final NodeManagerMetrics metrics; | ||
| private final OpportunisticContainersQueuePolicy oppContainersQueuePolicy; | ||
|
|
||
| private Boolean usePauseEventForPreemption = false; | ||
|
|
||
| private static int getMaxOppQueueLengthFromConf(final Context context) { | ||
| if (context == null || context.getConf() == null) { | ||
| return YarnConfiguration | ||
| .DEFAULT_NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH; | ||
| } | ||
|
|
||
| return context.getConf().getInt( | ||
| YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH, | ||
| YarnConfiguration.DEFAULT_NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH | ||
| ); | ||
| } | ||
|
|
||
| private static OpportunisticContainersQueuePolicy | ||
| getOppContainersQueuePolicyFromConf(final Context context) { | ||
|
||
| final String queuePolicy; | ||
| if (context == null || context.getConf() == null) { | ||
| queuePolicy = YarnConfiguration | ||
| .DEFAULT_NM_OPPORTUNISTIC_CONTAINERS_QUEUE_POLICY; | ||
| } else { | ||
| queuePolicy = context.getConf().get( | ||
|
||
| YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_QUEUE_POLICY, | ||
| YarnConfiguration | ||
| .DEFAULT_NM_OPPORTUNISTIC_CONTAINERS_QUEUE_POLICY | ||
| ); | ||
| } | ||
|
|
||
| return OpportunisticContainersQueuePolicy.valueOf(queuePolicy); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| ResourceHandlerChain resourceHandlerChain = null; | ||
|
|
||
|
|
@@ -120,10 +153,9 @@ public class ContainerScheduler extends AbstractService implements | |
| */ | ||
| public ContainerScheduler(Context context, AsyncDispatcher dispatcher, | ||
| NodeManagerMetrics metrics) { | ||
| this(context, dispatcher, metrics, | ||
| getOppContainersQueuePolicyFromConf(context), | ||
| getMaxOppQueueLengthFromConf(context)); | ||
| } | ||
|
|
||
|
|
||
|
|
@@ -149,13 +181,35 @@ public void serviceInit(Configuration conf) throws Exception { | |
| @VisibleForTesting | ||
| public ContainerScheduler(Context context, AsyncDispatcher dispatcher, | ||
| NodeManagerMetrics metrics, int qLength) { | ||
| this(context, dispatcher, metrics, | ||
| getOppContainersQueuePolicyFromConf(context), qLength); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| public ContainerScheduler(Context context, AsyncDispatcher dispatcher, | ||
| NodeManagerMetrics metrics, | ||
| OpportunisticContainersQueuePolicy oppContainersQueuePolicy, | ||
| int qLength) { | ||
| super(ContainerScheduler.class.getName()); | ||
| this.context = context; | ||
| this.dispatcher = dispatcher; | ||
| this.metrics = metrics; | ||
| this.utilizationTracker = | ||
| new AllocationBasedResourceUtilizationTracker(this); | ||
| this.oppContainersQueuePolicy = oppContainersQueuePolicy; | ||
| switch (oppContainersQueuePolicy) { | ||
| case BY_RESOURCES: | ||
| this.maxOppQueueLength = 0; | ||
| this.forceStartGuaranteedContainers = false; | ||
| LOG.info("Setting max opportunistic queue length to 0," | ||
| + " as {} is incompatible with queue length", | ||
| oppContainersQueuePolicy); | ||
| break; | ||
| case BY_QUEUE_LEN: | ||
| default: | ||
| this.maxOppQueueLength = qLength; | ||
| this.forceStartGuaranteedContainers = (maxOppQueueLength <= 0); | ||
| } | ||
| this.opportunisticContainersStatus = | ||
| OpportunisticContainersStatus.newInstance(); | ||
| } | ||
|
|
@@ -187,7 +241,7 @@ public void handle(ContainerSchedulerEvent event) { | |
| shedQueuedOpportunisticContainers(); | ||
| break; | ||
| case RECOVERY_COMPLETED: | ||
| startPendingContainers(forceStartGuaranteedContainers); | ||
| metrics.setQueuedContainers(queuedOpportunisticContainers.size(), | ||
| queuedGuaranteedContainers.size()); | ||
| break; | ||
|
|
@@ -243,7 +297,7 @@ private void onUpdateContainer(UpdateContainerSchedulerEvent updateEvent) { | |
| LOG.warn(String.format("Could not update resources on " + | ||
| "continer update of %s", containerId), ex); | ||
| } | ||
| startPendingContainers(forceStartGuaranteedContainers); | ||
| metrics.setQueuedContainers(queuedOpportunisticContainers.size(), | ||
| queuedGuaranteedContainers.size()); | ||
| } | ||
|
|
@@ -371,7 +425,6 @@ private void onResourcesReclaimed(Container container) { | |
| ExecutionType.OPPORTUNISTIC) { | ||
| this.metrics.completeOpportunisticContainer(container.getResource()); | ||
| } | ||
| startPendingContainers(forceStartGuaranteedContainers); | ||
| } | ||
| this.metrics.setQueuedContainers(queuedOpportunisticContainers.size(), | ||
|
|
@@ -380,13 +433,13 @@ private void onResourcesReclaimed(Container container) { | |
|
|
||
| /** | ||
| * Start pending containers in the queue. | ||
| * @param forceStartGContainers When this is true, start guaranteed | ||
| * container without looking at available resource | ||
| */ | ||
| private void startPendingContainers(boolean forceStartGContainers) { | ||
| // Start guaranteed containers that are paused, if resources available. | ||
| boolean resourcesAvailable = startContainers( | ||
| queuedGuaranteedContainers.values(), forceStartGContainers); | ||
| // Start opportunistic containers, if resources available. | ||
| if (resourcesAvailable) { | ||
| startContainers(queuedOpportunisticContainers.values(), false); | ||
|
|
@@ -429,6 +482,21 @@ private boolean resourceAvailableToStartContainer(Container container) { | |
| return this.utilizationTracker.hasResourcesAvailable(container); | ||
| } | ||
|
|
||
| private boolean resourceAvailableToQueueOppContainer( | ||
|
Contributor
There was a problem hiding this comment. IIUC, we are doing this check after localization is done. Shouldn't this check be done at the time of container start to avoid wasting NM resources.
Contributor
Author
There was a problem hiding this comment. IIRC I didn't do that since it's much more complicated and requires the
Contributor
There was a problem hiding this comment. I checked code and looks like it is same case with existing code also. I think there is scope of improvement there and that can be handle as separate jira. @goiri what do you think ?
Member
There was a problem hiding this comment. For this JIRA we can leave it as is for consistency with the rest. |
||
| Container newOppContainer) { | ||
| final Resource cumulativeResource = Resource.newInstance(Resources.none()); | ||
| for (final Container container : queuedGuaranteedContainers.values()) { | ||
| Resources.addTo(cumulativeResource, container.getResource()); | ||
| } | ||
|
|
||
| for (final Container container : queuedOpportunisticContainers.values()) { | ||
| Resources.addTo(cumulativeResource, container.getResource()); | ||
| } | ||
|
|
||
| Resources.addTo(cumulativeResource, newOppContainer.getResource()); | ||
| return this.utilizationTracker.hasResourcesAvailable(cumulativeResource); | ||
goiri marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| private boolean enqueueContainer(Container container) { | ||
| boolean isGuaranteedContainer = container.getContainerTokenIdentifier(). | ||
| getExecutionType() == ExecutionType.GUARANTEED; | ||
|
|
@@ -438,7 +506,21 @@ private boolean enqueueContainer(Container container) { | |
| queuedGuaranteedContainers.put(container.getContainerId(), container); | ||
| isQueued = true; | ||
| } else { | ||
| switch (oppContainersQueuePolicy) { | ||
| case BY_RESOURCES: | ||
| isQueued = resourceAvailableToQueueOppContainer(container); | ||
| break; | ||
| case BY_QUEUE_LEN: | ||
| default: | ||
| if (maxOppQueueLength <= 0) { | ||
| isQueued = false; | ||
| } else { | ||
| isQueued = | ||
| queuedOpportunisticContainers.size() < maxOppQueueLength; | ||
| } | ||
| } | ||
|
|
||
| if (isQueued) { | ||
| LOG.info("Opportunistic container {} will be queued at the NM.", | ||
| container.getContainerId()); | ||
| queuedOpportunisticContainers.put( | ||
|
|
@@ -451,7 +533,6 @@ private boolean enqueueContainer(Container container) { | |
| container.sendKillEvent( | ||
| ContainerExitStatus.KILLED_BY_CONTAINER_SCHEDULER, | ||
| "Opportunistic container queue is full."); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -484,7 +565,6 @@ protected void scheduleContainer(Container container) { | |
| // When opportunistic container not allowed (which is determined by | ||
| // max-queue length of pending opportunistic containers <= 0), start | ||
| // guaranteed containers without looking at available resources. | ||
| startPendingContainers(forceStartGuaranteedContainers); | ||
|
|
||
| // if the guaranteed container is queued, we need to preempt opportunistic | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.