ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3655481710
##########
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. The cap is now the executor's **actual registered total cores**, not
the ResourceProfile/configured cores. I added `WorkerOffer.totalCores`
(populated from `ExecutorData.totalCores` in `CoarseGrainedSchedulerBackend`,
and from the local backend's total cores), threaded it per-offer into
`effectiveCpusFor`, and cap the retry at `min(offer's actual total cores,
profile/configured cores)`:
```scala
val cap = (offerTotalCores.toSeq ++ profileCoresLimit.toSeq) match {
case Seq() => None
case caps => Some(caps.min)
}
```
So in the Kubernetes OOM-recovery case (replacement registers
`spark.task.cpus` cores) the retry is capped at what the executor actually has
and can still launch / reach `maxTaskFailures`, rather than being rejected on
every offer. Added unit coverage for a registered-total smaller than the
configured cores, and an end-to-end `FailureSuite` test with `local[2,2]` +
`spark.executor.cores=4` (unequal configured vs actual) that completes instead
of hanging.
Note: rebased onto master after [SPARK-58192] Support fractional
spark.task.cpus; all CPU accounting in this PR is now the normalized
`BigDecimal`/`CpuAmount` representation, with a fractional retry-growth
regression added (`spark.task.cpus=0.5`, increment `0.5`, grows 0.5 -> 1.0 ->
1.5 exactly).
--
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]
