LuciferYang commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3812118865
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,201 @@ 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 _: ExecutorAllocationClient => true
+ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * @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 b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ synchronized {
+ if (_executorsHeld) {
+ true
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume: prefer what the
application actually
+ // requested (e.g. through requestExecutors), which the backend
tracks. When it
+ // never pushed a requirement, Standalone has no explicit one 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.
+ // Use the tracked requirement only when it was explicitly
requested: killing an
+ // executor of a static application seeds it with a bookkeeping
zero, which is
+ // not a requirement to restore.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend if
cg.hasExplicitExecutorRequests =>
+ cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0)
Review Comment:
`heldNumExecutors` reads what the driver has pushed rather than what the
application currently
wants, and it is wrong in three ways. With dynamic allocation off that map
starts empty (YARN and K8s
set the initial target without going through the driver) and
`requestExecutors` accumulates from
zero, so an application with `spark.executor.instances=100` that ever called
`sc.requestExecutors(3)`
is recorded as 3 and resume pushes 3 while the UI still shows Running.
`requestExecutors(0)` is legal
too, and then resume pushes zero and the application is parked at no
executors. In the other
direction the absolute zero that `killExecutors` writes does not set the
flag, so line 2150's
fallback applies: on standalone resume pushes `Int.MaxValue`, which with
`spark.cores.max` unset
takes every free core and undoes the downscale the operator just performed.
This looks like the same root cause as the executor-requirement point above:
have the backend
maintain the application's desired target, written by both public APIs and
by `adjustExecutors`, and
let resume push that. The seed differs per cluster manager, and standalone
starts unbounded
(`initialExecutorLimit` is None with dynamic allocation off), so its
`killExecutors` case has to
derive the target from the live executor count — decrementing an unbounded
value gets nowhere.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2082,201 @@ 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 _: ExecutorAllocationClient => true
+ 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,
+ * while 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.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * @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 b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ synchronized {
+ if (_executorsHeld) {
+ true
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume: prefer what the
application actually
+ // requested (e.g. through requestExecutors), which the backend
tracks. When it
+ // never pushed a requirement, Standalone has no explicit one 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.
+ // Use the tracked requirement only when it was explicitly
requested: killing an
+ // executor of a static application seeds it with a bookkeeping
zero, which is
+ // not a requirement to restore.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend if
cg.hasExplicitExecutorRequests =>
+ cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0)
+ } else {
+ schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.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
+ b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.setExecutorsHeld(true)
+ case _ =>
+ }
+ executorAllocationManager match {
+ case Some(manager) =>
+ val acknowledged = manager.suspend()
+ drainHeldExecutors(b)
+ acknowledged
+ case None => zeroExecutorRequirementAndDrain(b)
+ }
+ }
+ }
+ 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,
Review Comment:
With dynamic allocation off, a barrier job submitted during a hold fails
after about ten minutes
instead of waiting for the resume the way an ordinary job does.
`isExecutorActive` excludes
`executorsPendingDecommission`, so `maxNumConcurrentTasks` reads 0 as soon
as `drainHeldExecutors`
runs, without waiting for any executor to exit;
`checkBarrierStageWithNumSlots` then throws
`BarrierJobSlotsNumberCheckFailed`, and DAGScheduler re-posts every 15s and
calls `jobFailed` after 40
failures. MLlib distributed training and TorchDistributor are in scope.
Not consuming that failure count while held (or resetting it on resume)
would let these jobs wait like
any other. If touching the scheduler is not wanted, saying in
`docs/web-ui.md` that barrier jobs fail
during a hold would also do.
##########
core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala:
##########
@@ -62,4 +66,44 @@ 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 def holdRequestExecutorPool: ExecutorService = synchronized {
+ holdRequestExecutor.getOrElse {
+ val pool =
ThreadUtils.newDaemonSingleThreadExecutor("spark-ui-hold-resume")
+ holdRequestExecutor = Some(pool)
+ pool
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ holdRequestExecutor.foreach(_.shutdownNow())
+ }
+
+ def handleHoldRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.execute(() =>
Utils.tryLogNonFatalError(ctx.holdExecutors()))
+ // Do a quick pause here to give Spark time to decommission the
executors so the
+ // application shows up as held after the refresh. Note that this will
block the
+ // serving thread so the time should be limited in duration.
+ Thread.sleep(100)
+ }
+ }
+ }
+
+ def handleResumeRequest(request: HttpServletRequest): Unit = {
+ if (holdEnabled &&
parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
+ sc.filter(_.executorHoldSupported).foreach { ctx =>
+ holdRequestExecutorPool.execute(() =>
Utils.tryLogNonFatalError(ctx.resumeExecutors()))
Review Comment:
`Utils.tryLogNonFatalError(ctx.holdExecutors())` and the resume equivalent
discard the `Boolean`, so
neither success nor failure of either operation shows up on the page and a
failure leaves only a
logWarning. There is a second layer: resume waits on the
`requestTotalExecutors` ask
(`spark.rpc.askTimeout`, 120s by default — the dynamic-allocation path
reaches the same `final
override` through `syncTargetsWithClient()`) and sets `_executorsHeld =
false` only afterwards, so the
reloaded page still reads "Held". With dynamic allocation on it can be twice
that: `EAM.resume()`
shares its monitor with the 100ms `schedule()`, which is itself blocked in
the same ask once a push
has been rejected.
`beforeRedirect` is `HttpServletRequest => Unit` and has no response to
write to, so the outcome has
to land in driver-side state the page reads — a `@volatile`
in-progress/failed state rendered next to
the Application line, for instance, which would also make the
`Thread.sleep(100)` unnecessary.
--
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]
