ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3614432740
##########
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))
+ }
+
+ // The number of CPUs a retry of the given task should request, given the
ResourceProfile's
+ // base taskCpus. For a task that has failed with OOM, this grows by
oomRetryCpusIncrement per
+ // OOM failure, capped at the executor total cores.
+ private def effectiveCpusFor(index: Int, baseCpus: Int): Int = {
+ val requested = baseCpus + oomRetryCpusIncrement * numOomRetries(index)
Review Comment:
Fixed in 8f657a78422. `effectiveCpusFor` now floors the result at `baseCpus`:
```scala
private def effectiveCpusFor(index: Int, baseCpus: Int): Int = {
val requested = baseCpus + oomRetryCpusIncrement * numOomRetries(index)
executorCoresLimit.map(_.min(requested)).getOrElse(requested).max(baseCpus)
}
```
Even when an extreme increment overflows the `Int` arithmetic to a negative
`requested`, the `.max(baseCpus)` floor guarantees a positive result, so
`oomRetryNeedsMoreCpus` and `prepareLaunchingTask` never see a non-positive
value and the `cpus > 0` assertion in `TaskDescription` is never reached with
scheduler state already mutated. The request degrades gracefully to `baseCpus`
in that case. Added a test (`maximum increment never yields a non-positive cpu
request`) that OOM-fails with `oomRetryCpusIncrement = Int.MaxValue` and
asserts the retry still launches with a positive cpu 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]
