dongjoon-hyun commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3813867682
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -739,6 +776,41 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
executorDataMap.keySet.toSeq
}
+ /** See `SparkContext.holdExecutors()`. */
+ private[spark] def setExecutorsHeld(held: Boolean): Unit = {
+ executorsHeld = held
Review Comment:
Fixed in 8bfca77 as suggested. The backend now owns a hold-aware publish
path: a new `publishTotals()` publishes the requested totals normally and
all-zero totals while held, and every publish site goes through it --
`requestExecutors`, `requestTotalExecutors`, `adjustExecutors`, `reset()`
(which now keeps the requested totals while held instead of clearing them), and
the registration-guard reassertion (which no longer overwrites the requested
totals). So requests made during a hold are recorded but nothing is allocated
until resume, and `resumeExecutors()` republishes the requested totals as-is
when any were explicitly made, before or during the hold. Added an interleaving
test covering both public request APIs: request while held is retained, a held
reassertion still publishes zero without touching it, and resume republishes it.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -175,6 +177,28 @@ private[spark] class ExecutorAllocationManager(
// (2) an executor idle timeout has elapsed.
@volatile private var initializing: Boolean = true
+ // Whether allocation is suspended because the executors are held. While
this is true,
+ // `schedule()` is a no-op so that pending tasks do not bring up new
executors.
+ // See `SparkContext.holdExecutors()`.
+ private var suspended: Boolean = false
+
+ // Whether the current executor targets still have to be pushed to the
cluster manager. Set
+ // when a push from `suspend()`/`resume()` fails or is rejected (e.g. before
the YARN AM has
+ // registered), and by `reset()`, which may run inside a cluster manager RPC
handler where a
+ // synchronous request would self-deadlock (e.g. YARN's
RegisterClusterManager). The push is
+ // performed from the allocation thread in `schedule()` and retried until
acknowledged, with
+ // an exponential backoff: the conditions it retries (an AM restart, an
unreachable cluster
Review Comment:
Fixed in 8bfca77 (reflowed across the two lines so the sentence stays
intact).
##########
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:
Partially fixed in 8bfca77. The killExecutors case is fixed as you
suggested: a kill-seeded requirement (non-empty but never explicitly requested)
now restores the live executor count, so resume neither parks the application
at zero nor undoes the downscale with an unbounded requirement. On the explicit
cases I read it differently: after `sc.requestExecutors(3)` (or an explicit
`requestTotalExecutors(0)`) the cluster-side target is already 3 (or 0) before
any hold -- the accumulate-from-zero bookkeeping is a pre-existing
`requestExecutors` behavior orthogonal to this PR -- so restoring the
explicitly requested total is a faithful restore of the pre-hold state rather
than a regression. That said, the direction you proposed is now in place: with
the hold-aware publish path from the review above, the backend maintains the
requested totals through both public APIs and `adjustExecutors`, and resume
republishes them (including requests made during the hold).
--
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]
