sunchao commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3606333887
##########
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] Preserve the ResourceProfile's base CPU request when executor cores are
unknown
For Standalone and local-cluster without an explicit `spark.executor.cores`,
`getExecutorCores` is intentionally empty because the executor can use the
worker's available cores, while `conf.get(EXECUTOR_CORES)` returns the generic
default of 1. Once the increment is enabled, `effectiveCpusFor` is called for
every attempt, including when `numOomRetries(index) == 0`, so a base
`spark.task.cpus=2` is reduced to one CPU. `TaskSchedulerImpl` prechecks the
offer against the two-CPU ResourceProfile request but later debits the one CPU
in `TaskDescription`, so an eight-core executor can launch seven nominally
two-CPU tasks instead of four; with the default one CPU per task, OOM retries
never grow at all. Please derive this cap from the executor's actual capacity
(or represent an unknown cap explicitly) and ensure `effectiveCpusFor` can
never return less than `baseCpus`.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -329,13 +354,19 @@ private[spark] class TaskSetManager(
execId: String,
host: String,
list: ArrayBuffer[Int],
- speculative: Boolean = false): Option[Int] = {
+ speculative: Boolean = false,
+ taskCpus: Int = sched.CPUS_PER_TASK,
+ availCpus: Int = Int.MaxValue): Option[Int] = {
var indexOffset = list.size
while (indexOffset > 0) {
indexOffset -= 1
val index = list(indexOffset)
+ // Strict wait for OOM retries: if a task that previously failed with
OOM needs more CPUs
+ // than this offer currently has free, leave it in the queue and look
for another task,
+ // rather than launching it with the same resources that already caused
the OOM.
if (!isTaskExcludededOnExecOrNode(index, execId, host) &&
- !(speculative && hasAttemptOnHost(index, host))) {
+ !(speculative && hasAttemptOnHost(index, host)) &&
+ !oomRetryNeedsMoreCpus(index, taskCpus, availCpus)) {
Review Comment:
[P1] Reserve capacity for enlarged retries across offer rounds
This condition leaves an enlarged retry queued when the current offer is too
small, but then continues scanning and may launch a normal one-CPU task. On a
saturated executor, `CoarseGrainedSchedulerBackend` makes a new offer
immediately after each individual task completion; each released core can
therefore be backfilled before two or more cores ever accumulate. The retry is
delayed until the normal backlog drains, and continuing work from competing
task sets can keep it waiting indefinitely. Dynamic allocation also derives
executor demand from the ResourceProfile's base `maxTasksPerExecutor`, so it
need not add an executor for this larger pending task. Please add a
reservation/draining mechanism across offer rounds and make allocation aware of
the enlarged task shape.
##########
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] Compute retry CPU growth without `Int` overflow
The new configuration accepts every nonnegative `Int`, so with `baseCpus =
1` and `oomRetryCpusIncrement = Int.MaxValue`, the first OOM retry wraps
`requested` to `Int.MinValue` before this clamp. `oomRetryNeedsMoreCpus` then
accepts the negative value, and `prepareLaunchingTask` mutates `copiesRunning`,
`taskInfos`, and the running-task set before `TaskDescription` asserts that
`cpus > 0`. The launch fails with scheduler state already recording a running
copy. Please calculate in `Long` and clamp safely before converting to `Int`,
or impose a configuration bound that makes the arithmetic provably safe.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala:
##########
@@ -426,23 +426,27 @@ private[spark] class TaskSchedulerImpl(
val prof =
sc.resourceProfileManager.resourceProfileFromId(taskSetRpID)
val taskCpus =
ResourceProfile.getTaskCpusOrDefaultForProfile(prof, conf)
val (taskDescOption, didReject, index) =
- taskSet.resourceOffer(execId, host, maxLocality, taskCpus,
taskResAssignments)
+ taskSet.resourceOffer(execId, host, maxLocality, taskCpus,
taskResAssignments,
+ availableCpus(i))
noDelayScheduleRejects &= !didReject
for (task <- taskDescOption) {
- val (locality, resources) = if (task != null) {
+ // The CPUs actually used may exceed the ResourceProfile's
taskCpus when the task is
+ // an OOM retry (see spark.task.oomRetryCpusIncrement); read it
from the
+ // TaskDescription so the availableCpus bookkeeping matches what
was launched.
+ val (locality, resources, cpusUsed) = if (task != null) {
tasks(i) += task
addRunningTask(task.taskId, execId, taskSet)
- (taskSet.taskInfos(task.taskId).taskLocality, task.resources)
+ (taskSet.taskInfos(task.taskId).taskLocality, task.resources,
task.cpus)
} else {
assert(taskSet.isBarrier, "TaskDescription can only be null
for barrier task")
val barrierTask = taskSet.barrierPendingLaunchTasks(index)
barrierTask.assignedOfferIndex = i
barrierTask.assignedCores = taskCpus
- (barrierTask.taskLocality, barrierTask.assignedResources)
+ (barrierTask.taskLocality, barrierTask.assignedResources,
taskCpus)
}
minLaunchedLocality = minTaskLocality(minLaunchedLocality,
Some(locality))
- availableCpus(i) -= taskCpus
+ availableCpus(i) -= cpusUsed
Review Comment:
[P2] Keep `LocalSchedulerBackend` free-core accounting in sync
This scheduler-side change correctly charges `TaskDescription.cpus`, but
`LocalSchedulerBackend` still subtracts and restores the fixed
`scheduler.CPUS_PER_TASK`. For example, in `local[4]` with
`spark.executor.cores=4`, base CPUs 1, and a two-CPU retry, launching that
retry plus two normal tasks consumes four CPUs here while the local endpoint
subtracts only three and advertises a phantom core. Another revive or task-set
submission can then oversubscribe the local executor. Please track each
launched task's actual CPU count in the local backend and restore that count on
completion.
##########
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] Preserve JVM-OOM classification when FAILED arrives before executor loss
`Executor.TaskRunner` sends an `ExceptionFailure` status before invoking the
fatal exception handler that exits with code 52. If the driver processes that
FAILED update first, `handleFailedTask` marks the attempt finished; its OOM
branch recognizes only `SparkOutOfMemoryError`, not
`java.lang.OutOfMemoryError`. When the subsequent `ExecutorExited(52, ...)`
reaches this block, `info.running` is already false, so the counter is never
incremented and the retry keeps its original CPU count. This is a normal
reachable ordering for the advertised JVM-heap-OOM path. Please classify fatal
`OutOfMemoryError` in the `ExceptionFailure` path or preserve enough attempt
state for the later definitive loss reason, and add a test with FAILED
preceding executor loss.
--
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]
