ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3627539943
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -125,6 +126,30 @@ private[spark] class TaskSetManager(
val successful = new Array[Boolean](numTasks)
private val numFailures = new Array[Int](numTasks)
+ // For each task, tracks the number of times it has failed due to
out-of-memory. Used to grow
+ // the number of CPUs allocated to a retry (see
spark.task.oomRetryCpusIncrement), which lowers
+ // the executor's concurrent task count and thus increases the retry's
execution-memory share.
+ // Reset to 0 when the task succeeds, mirroring numFailures.
+ private[scheduler] val numOomRetries = new Array[Int](numTasks)
+ private val oomRetryCpusIncrement = conf.get(config.OOM_RETRY_CPUS_INCREMENT)
+ // Executor exit codes treated as OOM (see
spark.task.oomRetryExecutorExitCodes); when an
+ // executor exits with one of these, the tasks it was running have their OOM
retry count bumped.
+ private val oomRetryExecutorExitCodes =
conf.get(config.OOM_RETRY_EXECUTOR_EXIT_CODES).toSet
+ // The executor total cores of this TaskSet's ResourceProfile. A
TaskSetManager is tied to a
+ // single ResourceProfile, so this is constant for its lifetime and caps the
OOM retry cpus.
+ private val executorCoresLimit: Int = {
+ val rp =
sched.sc.resourceProfileManager.resourceProfileFromId(taskSet.resourceProfileId)
+ rp.getExecutorCores.getOrElse(conf.get(EXECUTOR_CORES))
Review Comment:
Fixed in 136abaa041a. The cap is now treated as authoritative: when the
executor's total core count is unknown (no ResourceProfile cores and no
explicit `spark.executor.cores`), the OOM retry does not grow its cpus at all,
so it can never request more than the executor physically has and can never
become permanently unschedulable.
```scala
private def effectiveCpusFor(index: Int, baseCpus: Int): Int =
executorCoresLimit match {
case Some(cap) =>
val requested = baseCpus.toLong + oomRetryCpusIncrement.toLong *
numOomRetries(index)
requested.min(cap.toLong).max(baseCpus.toLong).toInt
case None =>
baseCpus // unknown capacity: do not grow, to avoid an unschedulable
over-request
}
```
This covers both cases you raised: the Standalone/local-cluster worker with
unknown cores no longer over-requests (it keeps base cpus and always launches /
can reach maxTaskFailures), and ordinary `local[4]` without an explicit
`spark.executor.cores` also keeps base cpus rather than being silently pinned.
The new test `OOM retry does not grow cpus when executor cores are unknown`
builds a TaskResourceProfile with `getExecutorCores == None`, asserts no
explicit `spark.executor.cores` is set, and verifies the retry stays at base
cpus and still launches on a 1-cpu offer (proving it does not hang). The doc
and configuration.md now state this behavior explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
