sunchao commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3627064390
##########
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:
[P2] The floor prevents the non-positive crash, but it does not preserve the
configured CPU-growth semantics.
With `baseCpus = 1`, `oomRetryCpusIncrement = Int.MaxValue`, and an executor
cap of 4, the first retry wraps to `Int.MinValue` and returns 1 CPU; the second
wraps to -1 and also returns 1; only the third wraps positive and is capped at
4. The documented formula would cap the first retry at 4. When the core cap is
unknown, that third retry can instead request 2147483646 CPUs and become
permanently unschedulable.
The new `>= 1` assertion verifies crash avoidance only, not the configured
behavior. Please compute in `Long` or use saturating arithmetic before applying
the cap, and assert the exact capped value in this test.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1247,6 +1318,11 @@ private[spark] class TaskSetManager(
// that the task is not running, and it is NetworkFailure rather
than TaskFailure.
case _ => !info.launching
}
+ // Grow the CPUs of the retry when a running task's executor died of a
JVM heap OOM. See
+ // the SparkOutOfMemoryError branch in handleFailedTask for the
non-fatal counterpart.
+ if (oomRetryCpusIncrement > 0 && !isBarrier && isOomExit &&
exitCausedByApp) {
Review Comment:
[P1] The direct `OutOfMemoryError` ordering is fixed, but wrapped OOMs still
bypass the retry growth.
`FileFormatDataWriter.enrichWriteError` catches any `Throwable` and wraps it
in a `SparkException`. `Executor` then constructs the `ExceptionFailure` from
that top-level wrapper, while `TaskSetManager.isOom` checks only the top-level
exception and class name. For a wrapped `SparkOutOfMemoryError`,
`Executor.isFatalError` reaches the cause but deliberately treats it as
nonfatal, so there is no executor-loss fallback and the retry keeps its
original CPU count. For a wrapped JVM `OutOfMemoryError`, the executor sends
the wrapper as `FAILED` before exiting; if the driver handles that update
first, the attempt becomes finished and the later OOM exit is skipped by the
`info.running` guard.
The current ordering tests use a top-level OOM and miss both cases. Please
carry an explicit OOM classification from the executor or inspect a bounded
preserved cause chain, and add coverage for wrapped `SparkOutOfMemoryError` and
wrapped JVM OOM.
--
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]
