ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3627541774
##########
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 136abaa041a. The growth is now computed in `Long` and clamped to
the cap before narrowing to `Int`, so the first retry is capped at the executor
cores exactly as documented rather than wrapping through negative values:
```scala
case Some(cap) =>
val requested = baseCpus.toLong + oomRetryCpusIncrement.toLong *
numOomRetries(index)
requested.min(cap.toLong).max(baseCpus.toLong).toInt
```
With `baseCpus = 1`, `oomRetryCpusIncrement = Int.MaxValue`, and cap 4, the
first retry now requests exactly 4 (not 1 after a wrap). The unknown-cap
over-request you noted is also gone: when the cap is unknown the retry does not
grow at all (see the reply on the capacity thread), so there is no
2147483646-cpu request. The test is rewritten to `maximum increment is capped
at executor cores without overflow` and asserts the exact capped value: a 3-cpu
offer must wait, and a 4-cpu offer launches with `cpus === 4`.
--
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]
