holdenk commented on a change in pull request #29817:
URL: https://github.com/apache/spark/pull/29817#discussion_r493940745
##########
File path: core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
##########
@@ -61,13 +61,34 @@ private[deploy] object DeployMessages {
}
/**
+ * An internal message that used by Master itself, in order to handle the
+ * `DecommissionWorkersOnHosts` request from `MasterWebUI` asynchronously.
+ * @param ids A collection of Worker ids, which should be decommissioned.
+ */
+ case class DecommissionWorkers(ids: Seq[String]) extends DeployMessage
+
+ /**
+ * A message that sent from Master to Worker to decommission the Worker.
+ * It's used for the case where decommission is triggered at MasterWebUI.
+ *
+ * Note that decommission a Worker will cause all the executors on that
Worker
+ * to be decommissioned as well.
+ */
+ object DecommissionWorker extends DeployMessage
+
+ /**
+ * A message that sent to the Worker itself when it receives PWR signal,
Review comment:
Do you mean "sent by the worker to itself"?
##########
File path: core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
##########
@@ -61,13 +61,34 @@ private[deploy] object DeployMessages {
}
/**
+ * An internal message that used by Master itself, in order to handle the
+ * `DecommissionWorkersOnHosts` request from `MasterWebUI` asynchronously.
+ * @param ids A collection of Worker ids, which should be decommissioned.
+ */
+ case class DecommissionWorkers(ids: Seq[String]) extends DeployMessage
+
+ /**
+ * A message that sent from Master to Worker to decommission the Worker.
+ * It's used for the case where decommission is triggered at MasterWebUI.
+ *
+ * Note that decommission a Worker will cause all the executors on that
Worker
+ * to be decommissioned as well.
+ */
+ object DecommissionWorker extends DeployMessage
+
+ /**
+ * A message that sent to the Worker itself when it receives PWR signal,
+ * indicating the Worker starts to decommission.
Review comment:
I'm not sure about this parts meaning
##########
File path:
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
##########
@@ -166,17 +171,6 @@ private[spark] class CoarseGrainedExecutorBackend(
if (executor == null) {
exitExecutor(1, "Received LaunchTask command but executor was null")
} else {
- if (decommissioned) {
- val msg = "Asked to launch a task while decommissioned."
- logError(msg)
- driver match {
- case Some(endpoint) =>
- logInfo("Sending DecommissionExecutor to driver.")
- endpoint.send(DecommissionExecutor(executorId,
ExecutorDecommissionInfo(msg)))
- case _ =>
- logError("No registered driver to send Decommission to.")
- }
- }
Review comment:
I don't think we should just take this out. async sends could fail,
re-sending the message if we receive a request which indicates the master
hasn't received our notification indicates we should resend.
##########
File path:
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
##########
@@ -272,10 +268,14 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
removeWorker(workerId, host, message)
context.reply(true)
- case DecommissionExecutor(executorId, decommissionInfo) =>
- logError(s"Received decommission executor message ${executorId}:
${decommissionInfo}.")
- context.reply(decommissionExecutor(executorId, decommissionInfo,
- adjustTargetNumExecutors = false))
+ case ExecutorDecommissioning(executorId) =>
+ logWarning(s"Received executor $executorId decommissioned message")
Review comment:
Here might be where you break the test suite last time, so double check
it.
##########
File path:
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
##########
@@ -213,9 +207,17 @@ private[spark] class CoarseGrainedExecutorBackend(
logInfo(s"Received tokens of ${tokenBytes.length} bytes")
SparkHadoopUtil.get.addDelegationTokens(tokenBytes, env.conf)
- case DecommissionSelf =>
- logInfo("Received decommission self")
+ case DecommissionExecutor =>
decommissionSelf()
+
+ case ExecutorSigPWRReceived =>
+ decommissionSelf()
+ if (driver.nonEmpty) {
Review comment:
Doing this after decommissionSelf introduces a race condition, not too
important one but given that it wasn't in the original flow I'd suggest
swapping the order.
##########
File path: core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
##########
@@ -70,7 +70,10 @@ private[deploy] class Worker(
if (conf.get(config.DECOMMISSION_ENABLED)) {
logInfo("Registering SIGPWR handler to trigger decommissioning.")
SignalUtils.register("PWR", "Failed to register SIGPWR handler - " +
- "disabling worker decommission feature.")(decommissionSelf)
+ "disabling worker decommission feature.") {
+ self.send(WorkerSigPWRReceived)
Review comment:
I think this might mean we return from handling the signal right away
rather than waiting for decommissionSelf to be finished. Is this an intentional
change?
Also this will no longer report a decommissioning failure by signal return
value, so may block pod deletion or other cleanup task longer than needed.
##########
File path: core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
##########
@@ -668,8 +672,13 @@ private[deploy] class Worker(
finishedApps += id
maybeCleanupApplication(id)
- case WorkerDecommission(_, _) =>
+ case DecommissionWorker =>
+ decommissionSelf()
+
+ case WorkerSigPWRReceived =>
decommissionSelf()
+ // Tell master we starts decommissioning so it stops trying to launch
executor/driver on us
Review comment:
May I suggest "Tell the master that we are starting decommissioning so
it stops trying to launch executor/driver on us"
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -1809,7 +1809,9 @@ private[spark] class BlockManager(
blocksToRemove.size
}
- def decommissionBlockManager(): Unit = synchronized {
+ def decommissionBlockManager(): Unit =
storageEndpoint.ask(DecommissionBlockManager)
Review comment:
Why?
##########
File path:
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
##########
@@ -213,9 +207,17 @@ private[spark] class CoarseGrainedExecutorBackend(
logInfo(s"Received tokens of ${tokenBytes.length} bytes")
SparkHadoopUtil.get.addDelegationTokens(tokenBytes, env.conf)
- case DecommissionSelf =>
- logInfo("Received decommission self")
+ case DecommissionExecutor =>
decommissionSelf()
+
+ case ExecutorSigPWRReceived =>
+ decommissionSelf()
+ if (driver.nonEmpty) {
+ // Tell driver we starts decommissioning so it stops trying to
schedule us
Review comment:
s/we starts/we are starting/
?
##########
File path:
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
##########
@@ -79,12 +79,17 @@ private[spark] class CoarseGrainedExecutorBackend(
*/
private[executor] val taskResources = new mutable.HashMap[Long, Map[String,
ResourceInformation]]
- @volatile private var decommissioned = false
+ private var decommissioned = false
override def onStart(): Unit = {
- logInfo("Registering PWR handler.")
- SignalUtils.register("PWR", "Failed to register SIGPWR handler - " +
- "disabling decommission feature.")(decommissionSelf)
+ if (env.conf.get(DECOMMISSION_ENABLED)) {
+ logInfo("Registering PWR handler to trigger decommissioning.")
+ SignalUtils.register("PWR", "Failed to register SIGPWR handler - " +
+ "disabling executor decommission feature.") {
+ self.send(ExecutorSigPWRReceived)
+ true
+ }
+ }
Review comment:
Similar comments as in Worker.scala
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
