[SPARK-9731] Standalone scheduling incorrect cores if spark.executor.cores is not set

[carsonwang]

The issue only happens if spark.executor.cores is not set and executor memory is set to a high value.
For example, if we have a worker with 4G and 10 cores and we set spark.executor.memory to 3G, then only 1 core is assigned to the executor. The correct number should be 10 cores.
I've added a unit test to illustrate the issue.

[carsonwang]

@andrewor14 , would you mind taking a look at this? This is related to recently changes in #7274 and #7668

[SparkQA]

Test build #40130 has finished for PR 8017 at commit 943cc4c.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[carsonwang]

retest this please

[srowen]

You guys probably understand this better than I, but I don't immediately see why this would change the behavior described in the JIRA. It allows executor scheduling where it didn't before (in the case where no executors are being launched though?) but why does that make more cores get assigned to the executor that's launched? I also couldn't see where the available memory was set to 4g.

[andrewor14]

Great catch!! I believe that was fixed in #7668 but was later introduced again in #7532. Thanks for adding the unit test.

[andrewor14]

I guess this is not affected because minCoresPerExecutor is 1 in this case.

[andrewor14]

retest this please. I'm upgrading this to a 1.5 blocker.

[andrewor14]

would you mind rewriting this as follows now that we have more shared code?

val keepScheduling = coresToAssign >= minCoresPerExecutor
val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor

// If we allow multiple executors per worker, then we can always launch new executors.
// Otherwise, if there is already an executor on this worker, just give it more cores.
val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutors(pos) == 0
if (launchingNewExecutor) {
  val assignedMemory = assignedExecutors(pos) * memoryPerExecutor
  val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
  val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
  keepScheduling && enoughCores && enoughMemory && underLimit
} else {
  // We're adding cores to an existing executor, so no need to check memory and executor limits
  keepScheduling && enoughCores
}

[carsonwang]

Looks great. I'll update it.

[SparkQA]

Test build #40178 has finished for PR 8017 at commit 943cc4c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #40219 has finished for PR 8017 at commit 86b651f.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[carsonwang]

retest this please.

[SparkQA]

Test build #40220 has finished for PR 8017 at commit d09ec48.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[andrewor14]

LGTM, merging into master and 1.5. Thanks for your fix.