[SPARK-4751] Dynamic allocation in standalone mode

core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala

@@ -177,16 +177,14 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
           s"${now - lastSeenMs} ms exceeds timeout $executorTimeoutMs ms")
         scheduler.executorLost(executorId, SlaveLost("Executor heartbeat " +
           s"timed out after ${now - lastSeenMs} ms"))
-        if (sc.supportDynamicAllocation) {
           // Asynchronously kill the executor to avoid blocking the current thread
-          killExecutorThread.submit(new Runnable {
-            override def run(): Unit = Utils.tryLogNonFatalError {
-              // Note: we want to get an executor back after expiring this one,
-              // so do not simply call `sc.killExecutor` here (SPARK-8119)
-              sc.killAndReplaceExecutor(executorId)
-            }
-          })
-        }
+        killExecutorThread.submit(new Runnable {
+          override def run(): Unit = Utils.tryLogNonFatalError {
+            // Note: we want to get an executor back after expiring this one,
+            // so do not simply call `sc.killExecutor` here (SPARK-8119)
+            sc.killAndReplaceExecutor(executorId)
+          }
+        })
         executorLastSeen.remove(executorId)
       }
     }

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -531,8 +531,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     val dynamicAllocationEnabled = _conf.getBoolean("spark.dynamicAllocation.enabled", false)
     _executorAllocationManager =
       if (dynamicAllocationEnabled) {
-        assert(supportDynamicAllocation,
-          "Dynamic allocation of executors is currently only supported in YARN and Mesos mode")
         Some(new ExecutorAllocationManager(this, listenerBus, _conf))
       } else {
         None
@@ -1361,17 +1359,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     postEnvironmentUpdate()
   }
 
-  /**
-   * Return whether dynamically adjusting the amount of resources allocated to
-   * this application is supported. This is currently only available for YARN
-   * and Mesos coarse-grained mode.
-   */
-  private[spark] def supportDynamicAllocation: Boolean = {
-    (master.contains("yarn")
-      || master.contains("mesos")
-      || _conf.getBoolean("spark.dynamicAllocation.testing", false))
-  }
-
   /**
    * :: DeveloperApi ::
    * Register a listener to receive up-calls from events that happen during execution.
@@ -1387,8 +1374,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
    * This is currently only supported in YARN mode. Return whether the request is received.
    */
   private[spark] override def requestTotalExecutors(numExecutors: Int): Boolean = {
-    assert(supportDynamicAllocation,
-      "Requesting executors is currently only supported in YARN and Mesos modes")
     schedulerBackend match {
       case b: CoarseGrainedSchedulerBackend =>
         b.requestTotalExecutors(numExecutors)
@@ -1405,8 +1390,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
    */
   @DeveloperApi
   override def requestExecutors(numAdditionalExecutors: Int): Boolean = {
-    assert(supportDynamicAllocation,
-      "Requesting executors is currently only supported in YARN and Mesos modes")
     schedulerBackend match {
       case b: CoarseGrainedSchedulerBackend =>
         b.requestExecutors(numAdditionalExecutors)
@@ -1429,8 +1412,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
    */
   @DeveloperApi
   override def killExecutors(executorIds: Seq[String]): Boolean = {
-    assert(supportDynamicAllocation,
-      "Killing executors is currently only supported in YARN and Mesos modes")
     schedulerBackend match {
       case b: CoarseGrainedSchedulerBackend =>
         b.killExecutors(executorIds)

core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala

@@ -107,6 +107,10 @@ private[deploy] object DeployMessages {
 
   case class MasterChangeAcknowledged(appId: String)
 
+  case class RequestExecutors(appId: String, requestedTotal: Int)
+
+  case class KillExecutors(appId: String, executorIds: Seq[String])
+
   // Master to AppClient
 
   case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage

core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala

@@ -70,6 +70,11 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana
     server = transportContext.createServer(port, bootstraps)
   }
 
+  /** Clean up all shuffle files associated with an application that has exited. */
+  def applicationRemoved(appId: String): Unit = {
+    blockHandler.applicationRemoved(appId, true /* cleanupLocalDirs */)
+  }
+
   def stop() {
     if (server != null) {
       server.close()

core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala

@@ -197,6 +197,22 @@ private[spark] class AppClient(
         sendToMaster(UnregisterApplication(appId))
         context.reply(true)
         stop()
+
+      case r: RequestExecutors =>
+        master match {
+          case Some(m) => context.reply(m.askWithRetry[Boolean](r))
+          case None =>
+            logWarning("Attempted to request executors before registering with Master.")
+            context.reply(false)
+        }
+
+      case k: KillExecutors =>
+        master match {
+          case Some(m) => context.reply(m.askWithRetry[Boolean](k))
+          case None =>
+            logWarning("Attempted to kill executors before registering with Master.")
+            context.reply(false)
+        }
     }
 
     override def onDisconnected(address: RpcAddress): Unit = {
@@ -256,4 +272,33 @@ private[spark] class AppClient(
       endpoint = null
     }
   }
+
+  /**
+   * Request executors from the Master by specifying the total number desired,
+   * including existing pending and running executors.
+   *
+   * @return whether the request is acknowledged.
+   */
+  def requestTotalExecutors(requestedTotal: Int): Boolean = {
+    if (endpoint != null && appId != null) {
+      endpoint.askWithRetry[Boolean](RequestExecutors(appId, requestedTotal))
+    } else {
+      logWarning("Attempted to request executors before driver fully initialized.")
+      false
+    }
+  }
+
+  /**
+   * Kill the given list of executors through the Master.
+   * @return whether the kill request is acknowledged.
+   */
+  def killExecutors(executorIds: Seq[String]): Boolean = {
+    if (endpoint != null && appId != null) {
+      endpoint.askWithRetry[Boolean](KillExecutors(appId, executorIds))
+    } else {
+      logWarning("Attempted to kill executors before driver fully initialized.")
+      false
+    }
+  }
+
 }

core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala

@@ -22,7 +22,6 @@ import java.util.Date
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.deploy.ApplicationDescription
 import org.apache.spark.rpc.RpcEndpointRef
 import org.apache.spark.util.Utils
@@ -43,6 +42,18 @@ private[spark] class ApplicationInfo(
   @transient var endTime: Long = _
   @transient var appSource: ApplicationSource = _
 
+  // A cap on the number of executors this application can have at any given time.
+  // By default, this is infinite. Only after the first allocation request is issued
+  // by the application will this be set to a finite value.
+  @transient var executorLimit: Int = _
+
+  // A set of workers on which this application cannot launch executors.
+  // This is used to handle kill requests when `spark.executor.cores` is NOT set. In this mode,
+  // at most one executor from this application can be run on each worker. When an executor is
+  // killed, its worker is added to the blacklist to avoid having the master immediately schedule
+  // a new executor on the worker.
+  @transient private var blacklistedWorkers: mutable.HashSet[String] = _
+
   @transient private var nextExecutorId: Int = _
 
   init()
@@ -60,6 +71,8 @@ private[spark] class ApplicationInfo(
     appSource = new ApplicationSource(this)
     nextExecutorId = 0
     removedExecutors = new ArrayBuffer[ExecutorDesc]
+    executorLimit = Integer.MAX_VALUE
+    blacklistedWorkers = new mutable.HashSet[String]
   }
 
   private def newExecutorId(useID: Option[Int] = None): Int = {
@@ -96,6 +109,47 @@ private[spark] class ApplicationInfo(
 
   private[master] def coresLeft: Int = requestedCores - coresGranted
 
+  /**
+   * Return the number of executors waiting to be scheduled once space frees up.
+   *
+   * This is only defined if the application explicitly set the executor limit. For instance,
+   * if an application asks for 8 executors but there is only space for 5, then there will be
+   * 3 waiting executors.
+   */
+  private[master] def numWaitingExecutors: Int = {
+    if (executorLimit != Integer.MAX_VALUE) {
+      math.max(0, executorLimit - executors.size)
+    } else {
+      0
+    }
+  }
+
+  /**
+   * Add a worker to the blacklist, called when the executor running on the worker is killed.
+   * This is used only if cores per executor is not set.
+   */
+  private[master] def blacklistWorker(workerId: String): Unit = {
+    blacklistedWorkers += workerId
+  }
+
+  /**
+   * Remove workers from the blacklist, called when the application requests new executors.
+   * This is used only if cores per executor is not set.
+   */
+  private[master] def removeFromBlacklist(numWorkers: Int): Unit = {
+    blacklistedWorkers.take(numWorkers).foreach { workerId =>
+      blacklistedWorkers.remove(workerId)
+    }
+  }
+
+  /**
+   * Return whether the specified worker is blacklisted.
+   * This is used only if cores per executor is not set.
+   */
+  private[master] def isBlacklisted(workerId: String): Boolean = {
+    blacklistedWorkers.contains(workerId)
+  }
+
   private var _retryCount = 0
 
   private[master] def retryCount = _retryCount