ulysses-you commented on code in PR #57329:
URL: https://github.com/apache/spark/pull/57329#discussion_r3614437303
##########
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:
Fixed in 8f657a78422. `LocalEndpoint` now tracks each launched task's actual
cpus in a `taskId -> cpus` map and refunds that exact count on completion,
instead of the fixed `CPUS_PER_TASK`:
```scala
private val runningTaskCpus = new mutable.HashMap[Long, Int]()
...
// StatusUpdate, on finished:
freeCores +=
runningTaskCpus.remove(taskId).getOrElse(scheduler.CPUS_PER_TASK)
...
// reviveOffers, per launched task:
freeCores -= task.cpus
runningTaskCpus(task.taskId) = task.cpus
```
Charging `task.cpus` at launch keeps the local free-core count consistent
with what `TaskSchedulerImpl` debited, so a two-cpu OOM retry no longer leaves
a phantom core that could oversubscribe the local executor. Added an end-to-end
test in `FailureSuite` that runs a `local[4,2]` job whose first attempts OOM
and whose retries need more cpus, asserting the job completes with the expected
task 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]
