sunchao commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3627063194
##########
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:
[P1] This still needs the executor actual total capacity, not just an
optional metadata cap.
The floor fixes the original under-request, but `None` now means no cap. On
a Standalone/local-cluster worker with four total cores, `baseCpus = 1` and an
increment of 4 make the first OOM retry request 5 CPUs. `oomRetryNeedsMoreCpus`
then rejects every offer permanently because `availCpus` can never exceed 4, so
the task cannot launch or reach `maxTaskFailures`. The new unit test supplies a
synthetic eight-CPU offer and therefore misses this case.
There is also a mirror problem in ordinary `local[4]`:
`ResourceProfile.getDefaultExecutorResources` takes the non-Standalone branch
and records the generic `spark.executor.cores` default of 1, while
`LocalSchedulerBackend` actually has four threads. Without explicitly setting
`spark.executor.cores=4`, retries remain capped at one CPU. The `FailureSuite`
test sets that property and masks the default behavior.
Please derive or validate the cap against actual executor capacity, and
cover both cases without overriding the offer/core count.
--
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]
