peter-toth commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3816122095
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,225 @@ 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 {
+ cg.requestTotalExecutors(
Review Comment:
**Finding 33.** The standalone sentinel is *recorded*, not just published,
and that changes `sc.killExecutors()` for the rest of the application's life.
This fallback goes through `cg.requestTotalExecutors`, which writes
`requestedTotalExecutorsPerResourceProfile(defaultProf) = Int.MaxValue` and
sets `explicitExecutorRequest = true`
(`core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:990-996`).
`adjustExecutors` branches on exactly that map being empty (`:1068`):
```scala
if (requestedTotalExecutorsPerResourceProfile.isEmpty) {
// Assume that we are killing an executor that was started by default and
// not through the request api
requestedTotalExecutorsPerResourceProfile(rp) = 0
} else {
requestedTotalExecutorsPerResourceProfile(rp) =
math.max(requestedTotalForRp - 1, 0)
}
```
Before a hold the map is empty, so killing an executor seeds 0 and the
Master stops replacing it. After a resume the map holds `Int.MaxValue`, so the
same kill publishes `Int.MaxValue - 1` and the Master replaces the executor
straight away — `sc.killExecutors()` quietly stops being a downscale.
Measured on `local-cluster[1,1,1024]` with the shuffle service and
decommissioning on, at `a927b1e`, control against treatment:
```
holdCycle=false initial totals={} explicit=false killSeededOnly=false
holdCycle=false afterKill killed=List(0) totals={rp0->0}
holdCycle=true initial totals={} explicit=false killSeededOnly=false
holdCycle=true afterResume totals={rp0->2147483647} explicit=true
killSeededOnly=false
holdCycle=true afterKill killed=List(1) totals={rp0->2147483646}
```
`explicit=true` on the third line is a second consequence:
`hasKillSeededTotalsOnly` is `!explicitExecutorRequest && nonEmpty`, so it can
never be true again and a *later* hold no longer restores `activeExecutorCount`
— it falls back to the unbounded branch. The kill-seeded restore path this PR
added is single-use.
You turned down "stop routing `Int.MaxValue` through
`requestTotalExecutors`" at finding 27 and you were right:
`ApplicationInfo.requestExecutors` iterates the map, so publishing an empty one
leaves the Master's limit at the 0 the hold set. But that objection is about
the *publish*, and it doesn't require *recording*. Publishing the sentinel
without recording it restores the exact pre-hold state — empty map, flag clear
— which is what "restore standalone's default" actually means:
```scala
/** Publish the given totals once, without recording them, and await the
acknowledgment. */
private[spark] def publishTotalsWithoutRecording(
totals: Map[ResourceProfile, Int]): Boolean = {
val response = synchronized { doRequestTotalExecutors(totals) }
defaultAskTimeout.awaitResult(response)
}
```
used in place of the `cg.requestTotalExecutors(...)` fallback here. It also
turns the `v != Int.MaxValue` guard you added to `updateExecRequestTimes` and
the saturating add at `requestExecutors:948` into defensive code rather than
something the feature depends on.
Scope so you can weigh it: standalone only — YARN and K8s restore a finite
`spark.executor.instances`, where decrementing is arguably the right answer —
and `killExecutors` requires dynamic allocation off. But the trigger is one
click of a control that ships on by default, and the failure is silent.
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +66,87 @@ private[ui] class JobsTab(parent: SparkUI, store:
AppStatusStore)
}
}
}
+
+ // Serves the hold/resume requests off the Jetty serving thread: they talk
to the cluster
+ // manager and may block up to the RPC ask timeout. A single thread also
serializes
+ // concurrent requests. Created on first use and shut down by `stop()`, so
that the thread
+ // does not outlive the SparkContext.
+ private var holdRequestExecutor: Option[ExecutorService] = None
+ private var stopped = false
+
+ // None once stopped, so that a request served during teardown neither hits
a rejected
+ // execution on the shut-down pool nor recreates it and leaks the thread.
+ private def holdRequestExecutorPool: Option[ExecutorService] = synchronized {
+ if (stopped) {
+ None
+ } else {
+ Some(holdRequestExecutor.getOrElse {
+ val pool =
ThreadUtils.newDaemonSingleThreadExecutor("spark-ui-hold-resume")
+ holdRequestExecutor = Some(pool)
+ pool
+ })
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ stopped = true
+ holdRequestExecutor.foreach(_.shutdownNow())
+ }
+
+ // Outcome of the last hold/resume request served by this tab, as (isHold,
message): Some
+ // while a request is running or after it did not take effect, None when
idle or after a
+ // success. AllJobsPage renders it only next to the matching control, so a
stale message
+ // never shows against the opposite operation.
+ @volatile private var holdRequestStatus: Option[(Boolean, String)] = None
+
+ private[jobs] def lastHoldRequestStatus(isHold: Boolean): Option[String] =
+ holdRequestStatus.collect { case (`isHold`, message) => message }
Review Comment:
**Finding 28.** The tagging fixed the mislabeling, but it also means a
failed hold can no longer be reported at all.
`holdExecutors()` sets `_executorsHeld = true` *before* it talks to the
cluster manager
(`core/src/main/scala/org/apache/spark/SparkContext.scala:2185`), deliberately
— "so that a partial failure below leaves the executors held". So when the
publish is rejected and `holdExecutors()` returns `false`, the page already
renders `Held … (resume)`, `action` is `"resume"`, and
`lastHoldRequestStatus(isHold = true)` returns `None`. `"the last hold request
did not take effect, see the driver logs"` is unreachable on exactly the path
it describes.
The three `false` returns:
- unsupported backend — unreachable from the UI anyway, `handleHoldRequest`
filters on `executorHoldSupported` first;
- a pipelined job is running — `_executorsHeld` stays `false`, so the
message does show, and this is the case where the generic wording says least;
- publish rejected, static or with dynamic allocation — message dropped.
Before the fix the message showed next to the wrong control; now it doesn't
show. Keeping the tag is right; what it shouldn't do is gate rendering on the
current state. Render it whenever it's set, labelled with the operation it
belongs to:
```scala
private[jobs] def lastHoldRequestStatus: Option[String] =
holdRequestStatus.map { case (isHold, message) =>
s"(${if (isHold) "hold" else "resume"}: $message)"
}
```
and drop the `isHold = action == "hold"` argument at `AllJobsPage.scala:378`.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2066,6 +2073,22 @@ 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 34.** The deferral returns before the job is registered, so the
job can't be cancelled while it waits.
`handleJobCancellation` does nothing when `jobIdToStageIds` has no entry for
the id (`:4317`), and that map is only populated from `submitStage`, which this
`return` skips. `doCancelAllJobs` walks `runningStages` (`:1654`) and
`cleanUpAfterSchedulerStop` walks `activeJobs` (`:2000`), so a deferred job is
in none of the three. No `SparkListenerJobStart` is posted either, so nothing
appears on the Jobs page.
Net effect: submit a pipelined job while held and you get a `collect()` that
hangs with no job visible in the UI, and `sc.cancelJob(id)` /
`sc.cancelAllJobs()` return without doing anything. It ends only on a resume or
`sc.stop()`. Cancelling the *job group* does work, because `handleJobSubmitted`
checks `cancelledJobGroups` at the top — that's the workaround, but nothing
tells a user about it.
For the barrier path this is a change from base rather than only a new gap:
base counts the failure, so the job fails after
`spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures` retries and the
waiter completes. Freezing the counter removes the only exit that existed.
Cheapest fix that keeps the "no partial scheduler state" property:
```scala
private[scheduler] val deferredWhileHeldJobIds = new HashSet[Int]
```
add the id before the re-post, remove it when the submission is
re-processed, and check it at the top of `handleJobCancellation` to fail the
listener and drop the id; the scheduled `Runnable` then sees the id gone and
skips the post. Same treatment covers the barrier branch at `:2135`.
If you'd rather not carry that state, then say it in the `holdExecutors()`
scaladoc: a barrier or pipelined job submitted during a hold waits for the
resume and cannot be cancelled on its own — cancel its job group instead.
--
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]
