Fix our exiting and cleanup thread for better debugging next time. Cleanup the locks we use in decommissioning and clarify some more bits.

Author: holdenk

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -285,39 +285,43 @@ private[spark] class CoarseGrainedExecutorBackend(
       // is viewed as acceptable to minimize introduction of any new locking structures in critical
       // code paths.
 
-      val shutdownThread = new Thread() {
-        var lastTaskRunningTime = System.nanoTime()
-        val sleep_time = 1000 // 1s
-
-        while (true) {
-          logInfo("Checking to see if we can shutdown.")
-          if (executor == null || executor.numRunningTasks == 0) {
-            if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
-              logInfo("No running tasks, checking migrations")
-              val allBlocksMigrated = env.blockManager.lastMigrationInfo()
-              // We can only trust allBlocksMigrated boolean value if there were no tasks running
-              // since the start of computing it.
-              if (allBlocksMigrated._2 &&
-                (allBlocksMigrated._1 > lastTaskRunningTime)) {
-                logInfo("No running tasks, all blocks migrated, stopping.")
-                exitExecutor(0, "Finished decommissioning", notifyDriver = true)
+      val shutdownExec = ThreadUtils.newDaemonSingleThreadExecutor("wait for decommissioning")
+      val shutdownRunnable = new Runnable() {
+        override def run(): Unit = {
+          var lastTaskRunningTime = System.nanoTime()
+          val sleep_time = 1000 // 1s
+
+          while (true) {
+            logInfo("Checking to see if we can shutdown.")
+            if (executor == null || executor.numRunningTasks == 0) {
+              if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
+                logInfo("No running tasks, checking migrations")
+                val allBlocksMigrated = env.blockManager.lastMigrationInfo()
+                // We can only trust allBlocksMigrated boolean value if there were no tasks running
+                // since the start of computing it.
+                if (allBlocksMigrated._2 &&
+                  (allBlocksMigrated._1 > lastTaskRunningTime)) {
+                  logInfo("No running tasks, all blocks migrated, stopping.")
+                  exitExecutor(0, "Finished decommissioning", notifyDriver = true)
+                } else {
+                  logInfo("All blocks not yet migrated.")
+                }
               } else {
-                logInfo("All blocks not yet migrated.")
+                logInfo("No running tasks, no block migration configured, stopping.")
+                exitExecutor(0, "Finished decommissioning", notifyDriver = true)
               }
+              Thread.sleep(sleep_time)
             } else {
-              logInfo("No running tasks, no block migration configured, stopping.")
-              exitExecutor(0, "Finished decommissioning", notifyDriver = true)
+              logInfo("Blocked from shutdown by running task")
+              // If there is a running task it could store blocks, so make sure we wait for a
+              // migration loop to complete after the last task is done.
+              Thread.sleep(sleep_time)
+              lastTaskRunningTime = System.nanoTime()
             }
-            Thread.sleep(sleep_time)
-          } else {
-            logInfo("Blocked from shutdown by running task")
-            // If there is a running task it could store blocks, so make sure we wait for a
-            // migration loop to complete after the last task is done.
-            Thread.sleep(sleep_time)
-            lastTaskRunningTime = System.nanoTime()
           }
         }
       }
+      shutdownExec.submit(shutdownRunnable)
       logInfo("Will exit when finished decommissioning")
       // Return true since we are handling a signal
       true

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -461,16 +461,14 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   override def decommissionExecutors(executorIds: Seq[String],
     adjustTargetNumExecutors: Boolean): Seq[String] = {
 
-    withLock {
+    CoarseGrainedSchedulerBackend.this.synchronized {
       val executorsToDecommission = executorIds.filter{executorId =>
-        CoarseGrainedSchedulerBackend.this.synchronized {
-          // Only bother decommissioning executors which are alive.
-          if (isExecutorActive(executorId)) {
-            executorsPendingDecommission += executorId
-            true
-          } else {
-            false
-          }
+        // Only bother decommissioning executors which are alive.
+        if (isExecutorActive(executorId)) {
+          executorsPendingDecommission += executorId
+          true
+        } else {
+          false
         }
       }
 
@@ -513,13 +511,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     }
     // Send decommission message to the executor (it could have originated on the executor
     // but not necessarily.
-    executorDataMap.get(executorId) match {
-      case Some(executorInfo) =>
-        executorInfo.executorEndpoint.send(DecommissionSelf)
-      case None =>
-        // Ignoring the executor since it is not registered.
-        logWarning(s"Attempted to decommission unknown executor $executorId.")
-        return false
+    CoarseGrainedSchedulerBackend.this.synchronized {
+      executorDataMap.get(executorId) match {
+        case Some(executorInfo) =>
+          executorInfo.executorEndpoint.send(DecommissionSelf)
+        case None =>
+          // Ignoring the executor since it is not registered.
+          logWarning(s"Attempted to decommission unknown executor $executorId.")
+          return false
+      }
     }
     logInfo(s"Finished decommissioning executor $executorId.")
 

core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala

@@ -43,9 +43,12 @@ class BlockManagerMaster(
     logInfo("Removed " + execId + " successfully in removeExecutor")
   }
 
-  /** Decommission block managers corresponding to given set of executors */
+  /** Decommission block managers corresponding to given set of executors
+   * Non-blocking.
+   */
   def decommissionBlockManagers(executorIds: Seq[String]): Unit = {
-    driverEndpoint.ask[Unit](DecommissionBlockManagers(executorIds))
+    driverEndpoint.ask[Boolean](DecommissionBlockManagers(executorIds))
+    logInfo(s"Decommissioning block managers on ${executorIds}")
   }
 
   /** Get Replication Info for all the RDD blocks stored in given blockManagerId */