Skip to content

[SPARK-21383][Core] Fix the YarnAllocator allocates more Resource #18651

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
libratiger wants to merge 8 commits into from
Closed

[SPARK-21383][Core] Fix the YarnAllocator allocates more Resource #18651

Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
818c912
[Core] Fix the YarnAllocator allocate more Resource
Jul 17, 2017
aba617e
[Core] Fix the YarnAllocator allocate more Resource
Jul 17, 2017
0980a6a
Fix the concurrent problem
Jul 19, 2017
856ebd6
Fix the merge conflict
Jul 19, 2017
bf7f78c
Remove unused code
Jul 19, 2017
cc73e5b
Add the debug log
Jul 19, 2017
582f424
make more robust
Jul 20, 2017
7703494
Fix the double count bug
Jul 25, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 34 additions & 26 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ private[yarn] class YarnAllocator(

@volatile private var numExecutorsRunning = 0

@volatile private var numExecutorToBeLaunched = 0

@vanzin vanzin Jul 18, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A better name would be numExecutorsStarting.

@libratiger libratiger Jul 19, 2017

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK，I will change the name.


/**
* Used to generate a unique ID per executor
*
Expand Down Expand Up @@ -294,7 +296,8 @@ private[yarn] class YarnAllocator(
def updateResourceRequests(): Unit = {
val pendingAllocate = getPendingAllocate
val numPendingAllocate = pendingAllocate.size
val missing = targetNumExecutors - numPendingAllocate - numExecutorsRunning
val missing = targetNumExecutors - numPendingAllocate -
numExecutorToBeLaunched - numExecutorsRunning

@tgravescs tgravescs Jul 19, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you also add in a debug message here, something like below. I found this very useful when debugging this issue and think it would be useful for debugging other allocating issues in the future.

logDebug(s"Updating resource requests, target: $targetNumExecutors, pending: " +

  •  s"$numPendingAllocate, running: $numExecutorsRunning, executorsStarting $numExecutorsPendingStart")

@libratiger libratiger Jul 19, 2017

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for your advice! I just add the debug info.

if (missing > 0) {
logInfo(s"Will request $missing executor container(s), each with " +
Expand Down Expand Up @@ -505,32 +508,37 @@ private[yarn] class YarnAllocator(

if (numExecutorsRunning < targetNumExecutors) {
if (launchContainers) {
launcherPool.execute(new Runnable {
override def run(): Unit = {
try {
new ExecutorRunnable(
Some(container),
conf,
sparkConf,
driverUrl,
executorId,
executorHostname,
executorMemory,
executorCores,
appAttemptId.getApplicationId.toString,
securityMgr,
localResources
).run()
updateInternalState()
} catch {
case NonFatal(e) =>
logError(s"Failed to launch executor $executorId on container $containerId", e)
// Assigned container should be released immediately to avoid unnecessary resource
// occupation.
amClient.releaseAssignedContainer(containerId)
try {
numExecutorToBeLaunched += 1

@vanzin vanzin Jul 18, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not thread safe. += is two operations.

launcherPool.execute(new Runnable {
override def run(): Unit = {
try {
new ExecutorRunnable(
Some(container),
conf,
sparkConf,
driverUrl,
executorId,
executorHostname,
executorMemory,
executorCores,
appAttemptId.getApplicationId.toString,
securityMgr,
localResources
).run()
updateInternalState()
} catch {
case NonFatal(e) =>
logError(s"Failed to launch executor $executorId on container $containerId", e)
// Assigned container should be released immediately
// to avoid unnecessary resource occupation.
amClient.releaseAssignedContainer(containerId)
}
}
}
})
})
} finally {
numExecutorToBeLaunched -= 1

@vanzin vanzin Jul 18, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here.

@vanzin vanzin Jul 18, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, shouldn't this be done in the actual thread that launching the executor? Otherwise you're decrementing the counter as soon as the task is submitted for execution, but you're not really waiting for the task to execute.

@libratiger libratiger Jul 19, 2017

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, you're right. When I test the code by experiment, i decrease the numExecutorToBeLaunched in the updateInternalState function, but I later found this may impact the test.

I will fix this soon.

}
} else {
// For test only
updateInternalState()
Expand Down