peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3819963818
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2143,6 +2151,23 @@ private[spark] class DAGScheduler(
return
}
+ // While the executors are held there are zero free slots by construction,
so the terminal
+ // gang admission check below would fail any pipelined job outright. Defer
the submission
+ // and re-check on a timer instead, like a barrier job: the shortfall is
not transient
+ // scarcity, and the job should wait for the resume.
+ if (hasPipelined && executorsHeld) {
Review Comment:
**Finding 35.** The deferral re-checks `executorsHeld`, but that is false
well before the cluster can admit the group again — so the re-post fails the
job outright.
`resumeExecutors()` publishes the restored requirement and returns; the
executors come back asynchronously. This PR's own test measures exactly that
gap: `assert(sc.resumeExecutors())` is followed by
`eventually(timeout(60.seconds)) { assert(sc.getExecutorIds().nonEmpty) }`
(`core/src/test/scala/org/apache/spark/SparkContextSuite.scala:1648-1652`). On
YARN or Kubernetes that wait is a queue wait or a pod schedule plus an image
pull, so in practice it is unbounded.
Meanwhile the re-post was armed up to `timeIntervalNumTasksCheck` (default
15s, `spark.scheduler.barrier.maxConcurrentTasksCheck.interval`) *before* the
resume, so it can fire anywhere in that gap. When it does, `executorsHeld` is
already false, this block is skipped, and `rejectUnadmittablePipelinedGroup`
sees `maxConcurrentTasksForProfile == 0` and calls `listener.jobFailed`
(`:1893-1902`) — terminal. There is no second attempt either, because
`deferredBarrierJobs.remove(jobId)` at `:2117` already dropped the entry.
Measured in `DAGSchedulerSuite` with `maxConcurrentTasksForTest = 0`,
re-posting the same `JobSubmitted` event the `messageScheduler` runnable posts
(substituting my own re-post for the timer is the one unmeasured link):
```
MEASURE35 held=true failed=false deferred=true stages=0
MEASURE35 held=false failed=true deferred=false stages=0 \
msg=[CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT] Cannot run the pipelined
stage group: it needs 4
concurrent task slots to run all its stages together, but only 0 are
currently free. ...
```
The barrier branch is the tell: it does not have this problem, because its
shortfall is retried up to
`spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures` times (40 x 15s
by default), which is what rides out a slow resume. The freeze you added at
`:2201` only stops the budget being *spent* during the hold; the budget itself
is what covers the warm-up afterwards. The pipelined path has no budget, so the
hold gave it a wait that ends one tick too early.
The state to fix it is already here: `handleJobSubmitted` can tell a re-post
from a first submission, because the `deferredBarrierJobs.remove(jobId)` at the
top returns non-null exactly for a re-post. Keep that value and let the
shortfall path decide defer-vs-fail, the way the barrier branch does:
```scala
// :2117 -- keep the entry instead of discarding it
val deferred = deferredBarrierJobs.remove(jobId)
if (deferred eq deferredJobCancelledMarker) {
return
}
...
// replacing the `hasPipelined && executorsHeld` block and the terminal
check below it:
// defer while held, and also while a job that was already deferred by a
hold still finds
// the group unadmittable -- the executors have not come back yet. Spend
the barrier budget
// once the hold is lifted, so a resume that never delivers executors
still fails the job
// instead of re-posting forever.
if (hasPipelined && !pipelinedGroupFits(jobId, finalRDD, partitions)) {
val stillWarmingUp = deferred != null && !executorsHeld &&
barrierJobIdToNumTasksCheckFailures.compute(jobId, (_, v) => v + 1)
<=
maxFailureNumTasksCheck
if (executorsHeld || stillWarmingUp) {
deferAndRepost(jobId, ...) // same put + messageScheduler.schedule
as today
return
}
failUnadmittablePipelinedGroup(jobId, ...) // today's
listener.jobFailed
return
}
```
That needs `rejectUnadmittablePipelinedGroup` split into a check and a
reject, which is a small refactor of `:1871-1906`. The alternative is to give
the pipelined shortfall the barrier's bounded retry unconditionally — the two
conditions really are the same "the cluster is temporarily too small", and this
PR has already decided the answer to it is to wait — but that changes pipelined
behaviour outside the hold, so it is your call whether that belongs here or in
a follow-up.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala:
##########
@@ -193,6 +193,17 @@ private[spark] class TaskSchedulerImpl(
}.sum
}
+ /**
+ * Whether any live task set belongs to a pipelined group. Its transient
shuffle data lives
+ * only on the executors, so it cannot survive holding them (see
+ * `SparkContext.holdExecutors()`).
+ */
Review Comment:
**Finding 36.** Nothing exercises `hasPipelinedTaskSets`, so the guard that
keeps a hold from aborting a running pipelined group is unverified.
`grep -rn hasPipelinedTaskSets core/src/test resource-managers` returns
nothing, and neither does the `holdExecutors()` branch it feeds
(`core/src/main/scala/org/apache/spark/SparkContext.scala:2145-2154`) — none of
the new `SparkContextSuite` tests submits a pipelined job. That branch is the
only thing between a hold and the group abort the scaladoc names ("a
force-killed member aborts the whole group"), and its failure mode is silence:
if the predicate stops matching — a future change to how zombie task sets are
tracked, or a pipelined task set reaching the scheduler by another route — the
hold just succeeds, and the group dies with `effectiveMaxTaskFailures = 1`
(`TaskSetManager.scala:80`).
Pinning the predicate is cheap; `TaskSetManagerSuite:3054-3058` already
builds a pipelined `TaskSet` the same way:
```scala
test("SPARK-58828: hasPipelinedTaskSets tracks live pipelined task sets") {
val taskScheduler = setupScheduler()
assert(!taskScheduler.hasPipelinedTaskSets)
taskScheduler.submitTasks(FakeTask.createTaskSet(1))
assert(!taskScheduler.hasPipelinedTaskSets)
val pipelined = new TaskSet(Array[Task[_]](new FakeTask(1, 0, Nil)),
stageId = 1,
stageAttemptId = 0, priority = 0, null,
ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID,
None, isPipelined = true)
taskScheduler.submitTasks(pipelined)
assert(taskScheduler.hasPipelinedTaskSets)
taskScheduler.taskSetManagerForAttempt(1, 0).get.isZombie = true
assert(!taskScheduler.hasPipelinedTaskSets)
}
```
The `holdExecutors()` branch itself needs a real pipelined job on a
`local-cluster`, which is a bigger ask — hence Non-blocking rather than
Blocking, since the submit-side deferral is covered and this is the second door.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,228 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors and can hold them,
decommission support,
+ * and shuffle data kept outside the executors: either an external shuffle
service or a
+ * `ShuffleDataIO` with reliable storage.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ (schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend => cg.supportsExecutorHold
+ case _ => false
+ }) &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
+ conf.get(DECOMMISSION_ENABLED)
+ }
+
+ /** Whether the executors are currently held by `holdExecutors()`. */
+ private[spark] def executorsHeld: Boolean = _executorsHeld
+
+ /**
+ * :: DeveloperApi ::
+ * Hold the whole application by declining to allocate new executors and
gracefully
+ * decommissioning all existing ones. Each executor finishes its running
tasks and then
+ * exits -- unless `spark.executor.decommission.forceKillTimeout` is set, in
which case an
+ * executor still running tasks is killed after that timeout. The shuffle
data already
+ * written remains available outside the executors, so the application can
later pick up
+ * where it left off via `resumeExecutors()`. Cached blocks are not
preserved and are
+ * recomputed after resuming. While held the application has no executors,
so with
+ * `spark.default.parallelism` unset the default parallelism falls back to
2, and an RDD
+ * created during the hold keeps that partition count after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`),
shuffle data kept
+ * outside the executors -- either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage -- and a
+ * scheduler backend that can hold executors: Standalone, YARN, and
Kubernetes with the
+ * `direct` pods allocator. Fallback storage
+ * (`spark.storage.decommission.fallbackStorage.path`) deliberately does not
qualify:
+ * shuffle blocks not yet migrated when an executor exits are dropped.
+ *
+ * Executor requirements requested while held, through `requestExecutors` or
+ * `requestTotalExecutors`, are recorded but nothing is allocated until
`resumeExecutors()`
+ * restores them.
+ *
+ * The hold is rejected, on a best-effort check, while a pipelined job is
running: its
+ * transient shuffle data lives only on the executors and would not survive
the drain. A
+ * group that slips past the check is aborted rather than drained, since a
pipelined task
+ * set tolerates no task failure. A pipelined job submitted while held waits
for the resume.
+ *
+ * @throws IllegalArgumentException when the decommission or shuffle-storage
precondition is
+ * not met; an unsupported scheduler backend instead returns false
with a warning.
+ * @return whether the lowered executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background; the
+ * executors are drained in either case.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend if cg.supportsExecutorHold =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ val pipelinedRunning = taskScheduler match {
+ case ts: TaskSchedulerImpl => ts.hasPipelinedTaskSets
+ case _ => false
+ }
+ if (pipelinedRunning) {
+ // A pipelined group reads and writes transient shuffle data that
lives only on
+ // its executors: a partially launched group would deadlock the
drain, and a
+ // force-killed member aborts the whole group.
+ logWarning(log"Cannot hold the executors while a pipelined job is
running.")
+ false
+ } else synchronized {
+ if (_executorsHeld) {
+ // A repeated hold re-asserts the zero requirement: the earlier
publish may not
+ // have been acknowledged, and with dynamic allocation off nothing
retries it.
+ if (executorAllocationManager.isDefined) true else
zeroExecutorRequirementAndDrain(cg)
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume when none was explicitly
requested
+ // (explicitly requested totals, made before or during the hold,
are
+ // republished from the backend directly). Only killExecutors'
bookkeeping
+ // zero is kill-seeded (read atomically against a concurrent
reset): restore
+ // the count of executors not already being removed, so that
resume neither
+ // parks the application at zero nor undoes the downscale.
Otherwise
+ // Standalone has no explicit requirement by default (and ignores
+ // spark.executor.instances, even a leftover value), so restore
an unbounded
+ // one; elsewhere follow the conf, or fall back to the cluster
manager's
+ // default when no executor has registered yet.
+ heldNumExecutors = if (cg.hasKillSeededTotalsOnly) {
+ cg.activeExecutorCount
+ } else {
+ schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(cg.getExecutorIds().size,
+ SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS))
+ }
+ }
+ }
+ // Mark the hold before talking to the cluster manager, so that a
partial failure
+ // below leaves the executors held, and thus resumable, instead of
half-held.
+ _executorsHeld = true
+ cg.setExecutorsHeld(true)
+ executorAllocationManager match {
+ case Some(manager) =>
+ val acknowledged = manager.suspend()
+ drainHeldExecutors(cg)
+ acknowledged
+ case None => zeroExecutorRequirementAndDrain(cg)
+ }
+ }
+ }
+ case _ =>
+ logWarning("Holding executors is not supported by current scheduler.")
+ false
+ }
+ }
+
+ // Gracefully decommission all the current executors of a held application
and let the
+ // executor monitor know, so that it does not try to remove the draining
executors again and
+ // reports them in the decommissioning metrics.
+ private def drainHeldExecutors(b: ExecutorAllocationClient): Unit = {
+ val executors = b.getExecutorIds()
+ if (executors.nonEmpty) {
+ val decommissioned = b.decommissionExecutors(
+ executors.map(id => (id, ExecutorDecommissionInfo("Executors are
held"))).toArray,
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
+ }
+ }
+
+ // Restore the hold invariant without dynamic allocation: publish the zero
requirement
+ // (the requested totals are kept and republished on resume) and drain the
current
+ // executors. The publish must not abort the drain, which has to run even
when the cluster
+ // manager is temporarily unreachable; the registration guard remains the
backstop when the
+ // publish fails.
+ private[spark] def zeroExecutorRequirementAndDrain(
+ cg: CoarseGrainedSchedulerBackend): Boolean = {
+ val acknowledged = try {
+ cg.republishRequestedTotals()
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to lower the executor requirement while holding
the " +
+ log"executors.", e)
+ false
+ }
+ drainHeldExecutors(cg)
+ acknowledged
+ }
+
+ /**
+ * :: DeveloperApi ::
+ * Resume an application held by `holdExecutors()` by restoring its executor
requirements.
+ *
+ * @return whether the restored executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background and the
+ * hold is lifted; otherwise the executors stay held so the call can
be retried.
+ */
+ @DeveloperApi
+ def resumeExecutors(): Boolean = {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend =>
+ synchronized {
+ if (!_executorsHeld) {
+ true
+ } else {
+ // Lift the backend guard before restoring the requirement, so
that an executor
+ // granted by the restored requirement cannot race with its own
registration and
+ // be drained.
+ cg.setExecutorsHeld(false)
+ val acknowledged = executorAllocationManager match {
+ case Some(manager) =>
+ // resume() retries a rejected push in the background, so the
hold can be
+ // lifted regardless of the acknowledgment.
+ manager.resume()
+ case None =>
+ try {
+ // Totals requested before or during the hold are
republished as-is; the
+ // check and the publish are atomic, so a concurrent cluster
manager reset
+ // cannot turn this into publishing an empty map. When none
are recorded
+ // (never requested, or cleared by such a reset), restore
the requirement
+ // captured at hold.
+ cg.republishExplicitTotals().getOrElse {
+ // Publish without recording: the pre-hold state had no
explicitly
+ // requested totals, and recording the restore (in
particular
+ // Standalone's unbounded sentinel) would flip
killExecutors' empty-map
+ // seeding and mark the totals explicit for good.
+ cg.publishTotalsWithoutRecording(
+
immutable.Map(resourceProfileManager.defaultResourceProfile ->
+ heldNumExecutors))
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to restore the executor requirement
while resuming " +
+ log"the executors.", e)
+ false
+ }
+ }
+ if (executorAllocationManager.isDefined || acknowledged) {
+ _executorsHeld = false
+ } else {
+ // The requirement could not be restored: stay held and re-arm
the guard. The
+ // rejected requestTotalExecutors recorded the non-zero
requirement before its
+ // acknowledgment failed, and an executor may have registered
while the guard
+ // was down, so restore the hold invariant: push the zero
requirement again and
+ // drain any current executors. The call can be retried.
Review Comment:
**Finding 37.** The comment describes a `requestTotalExecutors` call that
finding 33's fix removed.
Neither branch above records anything now: `republishExplicitTotals()`
republishes totals that were already recorded *before* the hold, and
`publishTotalsWithoutRecording()` records nothing by construction. So "the
rejected `requestTotalExecutors` recorded the non-zero requirement before its
acknowledgment failed" is no longer what happened. The conclusion still holds —
the recorded requirement really is non-zero, so pushing zero again is right —
but the reason is now "it was never lowered", not "the failed call raised it".
```suggestion
// The requirement could not be restored: stay held and re-arm
the guard. The
// recorded totals are still the non-zero pre-hold ones, and
an executor may
// have registered while the guard was down, so restore the
hold invariant:
// push the zero requirement again and drain any current
executors. The call
// can be retried.
```
--
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]
