dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3816275048
##########
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:
Fixed in 1614451 with `publishTotalsWithoutRecording`, exactly as sketched:
the non-explicit resume fallback publishes the restore (the Standalone sentinel
included) without touching the recorded totals or the explicit flag, so the
pre-hold state -- empty map, flag clear -- is what remains. `killExecutors`
keeps its empty-map seeding and a later hold keeps its kill-seeded restore,
closing both consequences you measured; the `updateExecRequestTimes` guard and
the saturating add are now defensive only. You are right that my finding-27
objection was about the publish, not the recording. Added a backend test
asserting the sentinel publish leaves the totals empty and the explicit record
clear.
##########
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:
Fixed in 1614451 with the deferred-jobs state, extended to carry the
listener (the id alone cannot fail the waiter): `jobsDeferredWhileHeld` maps
the deferred job id to its `JobListener`, both deferral sites register in it,
the re-post runnable skips when the id is gone (the plain barrier retry path is
unchanged), `handleJobSubmitted` removes the id on re-processing, and
`handleJobCancellation` fails a deferred listener directly. `doCancelAllJobs`
and `cleanUpAfterSchedulerStop` drain it too, so `cancelAllJobs()` and
`sc.stop()` complete the waiters as well. Added a test cancelling a deferred
pipelined job: the listener fails and the pending re-post is dropped.
--
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]
