diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index c8bd6410be27b..606f4700fff31 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -28,7 +28,7 @@ import org.apache.spark.metrics.GarbageCollectionMetrics import org.apache.spark.network.shuffle.Constants import org.apache.spark.network.shuffledb.DBBackend import org.apache.spark.network.util.ByteUnit -import org.apache.spark.scheduler.{EventLoggingListener, SchedulingMode} +import org.apache.spark.scheduler.{EventLoggingListener, SchedulingMode, TaskPlacementStrategy} import org.apache.spark.shuffle.sort.io.LocalDiskShuffleDataIO import org.apache.spark.storage.{DefaultTopologyMapper, RandomBlockReplicationPolicy} import org.apache.spark.unsafe.array.ByteArrayMethods @@ -2520,6 +2520,19 @@ package object config { .intConf .createWithDefault(5) + private[spark] val TASK_PLACEMENT_STRATEGY = + ConfigBuilder("spark.scheduler.taskPlacement.strategy") + .doc("The strategy used to traverse executor offers during the scheduler's NO_PREF and ANY " + + "passes. SPREAD cycles through shuffled eligible executor offers one task at a time. " + + "BIN_PACK fills each eligible executor before moving to the next. Executors with running " + + "or current-round assigned tasks are considered before idle executors; each group uses " + + "lexicographic executor ID order. The PROCESS_LOCAL, NODE_LOCAL, and RACK_LOCAL passes " + + "retain Spark's existing shuffled offer order.") + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .enumConf(TaskPlacementStrategy) + .createWithDefault(TaskPlacementStrategy.SPREAD) + private[spark] val SCHEDULER_REVIVE_INTERVAL = ConfigBuilder("spark.scheduler.revive.interval") .version("0.8.1") diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskPlacementStrategy.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskPlacementStrategy.scala new file mode 100644 index 0000000000000..22529a20bb7e2 --- /dev/null +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskPlacementStrategy.scala @@ -0,0 +1,24 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.scheduler + +private[spark] object TaskPlacementStrategy extends Enumeration { + type TaskPlacementStrategy = Value + + val SPREAD, BIN_PACK = Value +} diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index 618c8eb459026..dc7bb1cee0d20 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -183,6 +183,8 @@ private[spark] class TaskSchedulerImpl( // default scheduler is FIFO val schedulingMode: SchedulingMode = conf.get(SCHEDULER_MODE) + private val taskPlacementStrategy = conf.get(TASK_PLACEMENT_STRATEGY) + val rootPool: Pool = new Pool("", schedulingMode, 0, 0) // This is a var so that we can reset it for testing purposes. @@ -392,72 +394,94 @@ private[spark] class TaskSchedulerImpl( * @param maxLocality max locality to allow when scheduling * @param shuffledOffers shuffled resource offers to use for scheduling, * remaining resources are tracked by below fields as tasks are scheduled + * @param offerIndicesByExecutorId offer indices sorted by executor ID for BIN_PACK * @param availableCpus remaining cpus per offer, * value at index 'i' corresponds to shuffledOffers[i] * @param availableResources remaining resources per offer, * value at index 'i' corresponds to shuffledOffers[i] * @param tasks tasks scheduled per offer, value at index 'i' corresponds to shuffledOffers[i] + * @param useBinPacking whether to fill each offer before moving to the next one * @return tuple of (no delay schedule rejects?, option of min locality of launched task) */ private def resourceOfferSingleTaskSet( taskSet: TaskSetManager, maxLocality: TaskLocality, shuffledOffers: Seq[WorkerOffer], + offerIndicesByExecutorId: IndexedSeq[Int], availableCpus: Array[Int], availableResources: Array[ExecutorResourcesAmounts], - tasks: IndexedSeq[ArrayBuffer[TaskDescription]]) + tasks: IndexedSeq[ArrayBuffer[TaskDescription]], + useBinPacking: Boolean) : (Boolean, Option[TaskLocality]) = { var noDelayScheduleRejects = true var minLaunchedLocality: Option[TaskLocality] = None + val offerIndices = if (useBinPacking) { + // Re-evaluate busy executors for each TaskSet and bin-packed scheduling pass because tasks + // assigned by an earlier TaskSet or scheduling pass may have changed which executors should + // be packed first. Checking assigned cores also covers provisional barrier tasks, which are + // not registered as running until the entire barrier TaskSet can launch. + val (busyOfferIndices, idleOfferIndices) = offerIndicesByExecutorId.partition { i => + isExecutorBusy(shuffledOffers(i).executorId) || + availableCpus(i) < shuffledOffers(i).cores + } + busyOfferIndices.iterator ++ idleOfferIndices.iterator + } else { + shuffledOffers.indices.iterator + } // nodes and executors that are excluded for the entire application have already been // filtered out by this point - for (i <- shuffledOffers.indices) { - val execId = shuffledOffers(i).executorId - val host = shuffledOffers(i).host - val taskSetRpID = taskSet.taskSet.resourceProfileId - - // check whether the task can be scheduled to the executor base on resource profile. - if (sc.resourceProfileManager - .canBeScheduled(taskSetRpID, shuffledOffers(i).resourceProfileId)) { - val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, availableCpus(i), - availableResources(i)) - taskResAssignmentsOpt.foreach { taskResAssignments => - try { - val prof = sc.resourceProfileManager.resourceProfileFromId(taskSetRpID) - val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(prof, conf) - val (taskDescOption, didReject, index) = - taskSet.resourceOffer(execId, host, maxLocality, taskCpus, taskResAssignments) - noDelayScheduleRejects &= !didReject - for (task <- taskDescOption) { - val (locality, resources) = if (task != null) { - tasks(i) += task - addRunningTask(task.taskId, execId, taskSet) - (taskSet.taskInfos(task.taskId).taskLocality, task.resources) - } else { - assert(taskSet.isBarrier, "TaskDescription can only be null for barrier task") - val barrierTask = taskSet.barrierPendingLaunchTasks(index) - barrierTask.assignedOfferIndex = i - barrierTask.assignedCores = taskCpus - (barrierTask.taskLocality, barrierTask.assignedResources) + for (i <- offerIndices) { + var launchedTask = false + do { + launchedTask = false + val execId = shuffledOffers(i).executorId + val host = shuffledOffers(i).host + val taskSetRpID = taskSet.taskSet.resourceProfileId + + // check whether the task can be scheduled to the executor base on resource profile. + if (sc.resourceProfileManager + .canBeScheduled(taskSetRpID, shuffledOffers(i).resourceProfileId)) { + val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, availableCpus(i), + availableResources(i)) + taskResAssignmentsOpt.foreach { taskResAssignments => + try { + val prof = sc.resourceProfileManager.resourceProfileFromId(taskSetRpID) + val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(prof, conf) + val (taskDescOption, didReject, index) = + taskSet.resourceOffer(execId, host, maxLocality, taskCpus, taskResAssignments) + noDelayScheduleRejects &= !didReject + for (task <- taskDescOption) { + launchedTask = true + val (locality, resources) = if (task != null) { + tasks(i) += task + addRunningTask(task.taskId, execId, taskSet) + (taskSet.taskInfos(task.taskId).taskLocality, task.resources) + } else { + assert(taskSet.isBarrier, "TaskDescription can only be null for barrier task") + val barrierTask = taskSet.barrierPendingLaunchTasks(index) + barrierTask.assignedOfferIndex = i + barrierTask.assignedCores = taskCpus + (barrierTask.taskLocality, barrierTask.assignedResources) + } + + minLaunchedLocality = minTaskLocality(minLaunchedLocality, Some(locality)) + availableCpus(i) -= taskCpus + assert(availableCpus(i) >= 0) + availableResources(i).acquire(resources) } - - minLaunchedLocality = minTaskLocality(minLaunchedLocality, Some(locality)) - availableCpus(i) -= taskCpus - assert(availableCpus(i) >= 0) - availableResources(i).acquire(resources) + } catch { + case e: TaskNotSerializableException => + // scalastyle:off line.size.limit + logError(log"Resource offer failed, task set " + + log"${MDC(LogKeys.TASK_SET_NAME, taskSet.name)} was not serializable") + // scalastyle:on + // Do not offer resources for this task, but don't throw an error to allow other + // task sets to be submitted. + return (noDelayScheduleRejects, minLaunchedLocality) } - } catch { - case e: TaskNotSerializableException => - // scalastyle:off line.size.limit - logError(log"Resource offer failed, task set " + - log"${MDC(LogKeys.TASK_SET_NAME, taskSet.name)} was not serializable") - // scalastyle:on - // Do not offer resources for this task, but don't throw an error to allow other - // task sets to be submitted. - return (noDelayScheduleRejects, minLaunchedLocality) } } - } + } while (useBinPacking && launchedTask) } (noDelayScheduleRejects, minLaunchedLocality) } @@ -506,8 +530,9 @@ private[spark] class TaskSchedulerImpl( /** * Called by cluster manager to offer resources on workers. We respond by asking our active task - * sets for tasks in order of priority. We fill each node with tasks in a round-robin manner so - * that tasks are balanced across the cluster. + * sets for tasks in order of priority. During the NO_PREF and ANY scheduling passes, the + * configured task placement strategy determines whether tasks cycle through eligible offers or + * fill one executor before moving to the next. */ def resourceOffers( offers: IndexedSeq[WorkerOffer], @@ -553,6 +578,12 @@ private[spark] class TaskSchedulerImpl( val availableCpus = shuffledOffers.map(o => o.cores).toArray val resourceProfileIds = shuffledOffers.map(o => o.resourceProfileId).toArray val sortedTaskSets = rootPool.getSortedTaskSetQueue + val offerIndicesByExecutorId: IndexedSeq[Int] = + if (taskPlacementStrategy == TaskPlacementStrategy.BIN_PACK && sortedTaskSets.nonEmpty) { + shuffledOffers.indices.sortBy(i => shuffledOffers(i).executorId) + } else { + shuffledOffers.indices + } for (taskSet <- sortedTaskSets) { logDebug("parentName: %s, name: %s, runningTasks: %s".format( taskSet.parent.name, taskSet.name, taskSet.runningTasks)) @@ -588,11 +619,15 @@ private[spark] class TaskSchedulerImpl( var noDelaySchedulingRejects = true var globalMinLocality: Option[TaskLocality] = None for (currentMaxLocality <- taskSet.myLocalityLevels) { + val useBinPacking = + taskPlacementStrategy == TaskPlacementStrategy.BIN_PACK && + (currentMaxLocality == TaskLocality.NO_PREF || + currentMaxLocality == TaskLocality.ANY) var launchedTaskAtCurrentMaxLocality = false do { val (noDelayScheduleReject, minLocality) = resourceOfferSingleTaskSet( - taskSet, currentMaxLocality, shuffledOffers, availableCpus, - availableResources, tasks) + taskSet, currentMaxLocality, shuffledOffers, offerIndicesByExecutorId, + availableCpus, availableResources, tasks, useBinPacking) launchedTaskAtCurrentMaxLocality = minLocality.isDefined launchedAnyTask |= launchedTaskAtCurrentMaxLocality noDelaySchedulingRejects &= noDelayScheduleReject @@ -788,7 +823,8 @@ private[spark] class TaskSchedulerImpl( } /** - * Shuffle offers around to avoid always placing tasks on the same workers. Exposed to allow + * Shuffle offers to avoid fixed-order bias during SPREAD and locality-specific placement. + * BIN_PACK reorders them during the NO_PREF and ANY scheduling passes. Exposed to allow * overriding in tests, so it can be deterministic. */ protected def shuffleOffers(offers: IndexedSeq[WorkerOffer]): IndexedSeq[WorkerOffer] = { diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala index c9c9e529405ec..331dbfe063868 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala @@ -184,13 +184,13 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext assert(!scheduler.hasExecutorsAliveOnHost(host1)) } - test("Scheduler does not always schedule tasks on the same workers") { + test("default SPREAD placement does not always select the same executor") { val taskScheduler = setupScheduler() val numFreeCores = 1 val workerOffers = IndexedSeq(new WorkerOffer("executor0", "host0", numFreeCores), new WorkerOffer("executor1", "host1", numFreeCores)) - // Repeatedly try to schedule a 1-task job, and make sure that it doesn't always - // get scheduled on the same executor. While there is a chance this test will fail + // Repeatedly try to schedule a 1-task job, and make sure that default SPREAD placement + // doesn't always select the same executor. While there is a chance this test will fail // because the task randomly gets placed on the first executor all 1000 times, the // probability of that happening is 2^-1000 (so sufficiently small to be considered // negligible). @@ -208,6 +208,259 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext assert(!failedTaskSet) } + test("task placement strategy configuration") { + val conf = new SparkConf(false) + assert(conf.get(config.TASK_PLACEMENT_STRATEGY) === + TaskPlacementStrategy.SPREAD) + + conf.set(config.TASK_PLACEMENT_STRATEGY.key, "bin_pack") + assert(conf.get(config.TASK_PLACEMENT_STRATEGY) === TaskPlacementStrategy.BIN_PACK) + + conf.set(config.TASK_PLACEMENT_STRATEGY.key, "unsupported") + intercept[IllegalArgumentException] { + conf.get(config.TASK_PLACEMENT_STRATEGY) + } + } + + test("SPREAD task placement cycles a TaskSet across executor offers") { + val taskScheduler = setupScheduler() + val workerOffers = IndexedSeq( + WorkerOffer("executor3", "host3", 4), + WorkerOffer("executor1", "host1", 4), + WorkerOffer("executor2", "host2", 4), + WorkerOffer("executor0", "host0", 4)) + + taskScheduler.submitTasks(FakeTask.createTaskSet(8)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + val taskCounts = taskDescriptions.groupBy(_.executorId).map { + case (executorId, tasks) => executorId -> tasks.size + } + + assert(taskDescriptions.size === 8) + assert(taskCounts === Map( + "executor0" -> 2, + "executor1" -> 2, + "executor2" -> 2, + "executor3" -> 2)) + } + + test("BIN_PACK task placement fills executors in executor ID order") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + val workerOffers = IndexedSeq( + WorkerOffer("30", "host3", 4), + WorkerOffer("10", "host1", 4), + WorkerOffer("20", "host2", 4), + WorkerOffer("2", "host0", 4)) + + taskScheduler.submitTasks(FakeTask.createTaskSet(8)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + val taskCounts = taskDescriptions.groupBy(_.executorId).map { + case (executorId, tasks) => executorId -> tasks.size + } + + assert(taskDescriptions.size === 8) + assert(taskCounts === Map("10" -> 4, "2" -> 4)) + } + + test("BIN_PACK fills busy executors in executor ID order before idle executors") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + + taskScheduler.submitTasks(FakeTask.createTaskSet( + 1, stageId = 0, stageAttemptId = 0)) + val firstTask = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("20", "host20", 1))).flatten + assert(firstTask.map(_.executorId) === Seq("20")) + + taskScheduler.submitTasks(FakeTask.createTaskSet( + 1, stageId = 1, stageAttemptId = 0)) + val secondTask = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("10", "host10", 1))).flatten + assert(secondTask.map(_.executorId) === Seq("10")) + + taskScheduler.submitTasks(FakeTask.createTaskSet( + 7, stageId = 2, stageAttemptId = 0)) + val workerOffers = IndexedSeq( + WorkerOffer("20", "host20", 3), + WorkerOffer("00", "host00", 4), + WorkerOffer("10", "host10", 3)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + val taskCounts = taskDescriptions.groupBy(_.executorId).map { + case (executorId, tasks) => executorId -> tasks.size + } + + assert(taskDescriptions.size === 7) + assert(taskCounts === Map("10" -> 3, "20" -> 3, "00" -> 1)) + } + + test("BIN_PACK task placement preserves PROCESS_LOCAL placement") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + + taskScheduler.submitTasks(FakeTask.createTaskSet( + 1, stageId = 0, stageAttemptId = 0)) + val firstTask = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("executor1", "host1", 0), + WorkerOffer("executor0", "host0", 1))).flatten + assert(firstTask.map(_.executorId) === Seq("executor0")) + + val localTaskSet = FakeTask.createTaskSet( + 1, stageId = 1, stageAttemptId = 0, Seq(TaskLocation("host1", "executor1"))) + taskScheduler.submitTasks(localTaskSet) + val taskDescriptions = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("executor0", "host0", 3), + WorkerOffer("executor1", "host1", 1))).flatten + + assert(taskDescriptions.map(_.executorId) === Seq("executor1")) + val taskSetManager = taskScheduler.taskSetManagerForAttempt(1, 0).get + assert(taskSetManager.taskInfos(taskDescriptions.head.taskId).taskLocality === + TaskLocality.PROCESS_LOCAL) + } + + test("BIN_PACK task placement preserves NODE_LOCAL placement") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + + taskScheduler.submitTasks(FakeTask.createTaskSet( + 1, stageId = 0, stageAttemptId = 0)) + val firstTask = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("executor1", "host1", 0), + WorkerOffer("executor0", "host0", 1))).flatten + assert(firstTask.map(_.executorId) === Seq("executor0")) + + val localTaskSet = FakeTask.createTaskSet( + 1, stageId = 1, stageAttemptId = 0, Seq(TaskLocation("host1"))) + taskScheduler.submitTasks(localTaskSet) + val taskDescriptions = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("executor0", "host0", 3), + WorkerOffer("executor1", "host1", 1))).flatten + + assert(taskDescriptions.map(_.executorId) === Seq("executor1")) + val taskSetManager = taskScheduler.taskSetManagerForAttempt(1, 0).get + assert(taskSetManager.taskInfos(taskDescriptions.head.taskId).taskLocality === + TaskLocality.NODE_LOCAL) + } + + test("BIN_PACK prioritizes executors made busy by locality placement") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + val workerOffers = IndexedSeq( + WorkerOffer("executor0", "host0", 4), + WorkerOffer("executor9", "host9", 4)) + + taskScheduler.resourceOffers(workerOffers.map(_.copy(cores = 0))) + taskScheduler.submitTasks(FakeTask.createTaskSet( + 4, + Seq(TaskLocation("host9", "executor9")), + Nil, + Nil, + Nil)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + + assert(taskDescriptions.size === 4) + assert(taskDescriptions.forall(_.executorId == "executor9")) + } + + test("BIN_PACK reevaluates busy executors between TaskSets in one resource offer") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + val workerOffers = IndexedSeq( + WorkerOffer("executor0", "host0", 4), + WorkerOffer("executor9", "host9", 4)) + + taskScheduler.resourceOffers(workerOffers.map(_.copy(cores = 0))) + taskScheduler.submitTasks(FakeTask.createTaskSet( + 1, stageId = 0, stageAttemptId = 0, Seq(TaskLocation("host9", "executor9")))) + taskScheduler.submitTasks(FakeTask.createTaskSet( + 4, stageId = 1, stageAttemptId = 0)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + val taskCounts = taskDescriptions.groupBy(_.executorId).map { + case (executorId, tasks) => executorId -> tasks.size + } + + assert(taskDescriptions.size === 5) + assert(taskCounts === Map("executor9" -> 4, "executor0" -> 1)) + } + + test("BIN_PACK task placement is applied at ANY locality") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString, + config.LOCALITY_WAIT.key -> "0") + taskScheduler.resourceOffers( + IndexedSeq(WorkerOffer("executor9", "host9", 0))) + + val preferredLocations = + Seq.fill(4)(Seq(TaskLocation("host9", "executor9"))) + taskScheduler.submitTasks(FakeTask.createTaskSet(4, preferredLocations: _*)) + val taskDescriptions = taskScheduler.resourceOffers(IndexedSeq( + WorkerOffer("executor1", "host1", 4), + WorkerOffer("executor0", "host0", 4))).flatten + + assert(taskDescriptions.size === 4) + assert(taskDescriptions.forall(_.executorId == "executor0")) + val taskSetManager = taskScheduler.taskSetManagerForAttempt(0, 0).get + assert(taskSetManager.taskInfos.values.map(_.taskLocality).toSet === Set(TaskLocality.ANY)) + } + + test("BIN_PACK task placement respects custom resource limits") { + val taskScheduler = setupScheduler( + numCores = 4, + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString, + config.CPUS_PER_TASK.key -> "1", + TASK_GPU_ID.amountConf -> "1", + EXECUTOR_GPU_ID.amountConf -> "4", + config.EXECUTOR_CORES.key -> "4") + val workerOffers = IndexedSeq( + WorkerOffer("executor1", "host1", 4, None, + Map(GPU -> ArrayBuffer("2", "3", "4", "5"))), + WorkerOffer("executor0", "host0", 4, None, + Map(GPU -> ArrayBuffer("0", "1")))) + + taskScheduler.submitTasks(FakeTask.createTaskSet(4)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + val taskCounts = taskDescriptions.groupBy(_.executorId).map { + case (executorId, tasks) => executorId -> tasks.size + } + + assert(taskDescriptions.size === 4) + assert(taskCounts === Map("executor0" -> 2, "executor1" -> 2)) + } + + test("BIN_PACK task placement supports barrier tasks") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + val workerOffers = IndexedSeq( + WorkerOffer("executor1", "host1", 4, Some("192.168.0.101:49627")), + WorkerOffer("executor0", "host0", 4, Some("192.168.0.101:49625"))) + + taskScheduler.submitTasks(FakeTask.createBarrierTaskSet(4)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + + assert(taskDescriptions.size === 4) + assert(taskDescriptions.forall(_.executorId == "executor0")) + } + + test("BIN_PACK treats barrier locality assignments as busy") { + val taskScheduler = setupScheduler( + config.TASK_PLACEMENT_STRATEGY.key -> TaskPlacementStrategy.BIN_PACK.toString) + val workerOffers = IndexedSeq( + WorkerOffer("executor0", "host0", 4, Some("192.168.0.101:49625")), + WorkerOffer("executor9", "host9", 4, Some("192.168.0.101:49627"))) + + taskScheduler.resourceOffers(workerOffers.map(_.copy(cores = 0))) + taskScheduler.submitTasks(FakeTask.createBarrierTaskSet( + 4, + Seq(TaskLocation("host9", "executor9")), + Nil, + Nil, + Nil)) + val taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + + assert(taskDescriptions.size === 4) + assert(taskDescriptions.forall(_.executorId == "executor9")) + } + test("Scheduler correctly accounts for multiple CPUs per task") { val taskCpus = 2 val taskScheduler = setupSchedulerWithMaster( diff --git a/docs/configuration.md b/docs/configuration.md index 4999b08908dec..d56e379bfb200 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2852,6 +2852,20 @@ Apart from these, the following properties are also available, and may be useful 0.8.0 + + spark.scheduler.taskPlacement.strategy + SPREAD + + The strategy used to traverse executor offers during the scheduler's NO_PREF and + ANY passes. + SPREAD cycles through shuffled eligible executor offers one task at a time. + BIN_PACK fills each eligible executor before moving to the next. Executors with + running or current-round assigned tasks are considered before idle executors; each group uses + lexicographic executor ID order. The PROCESS_LOCAL, NODE_LOCAL, and + RACK_LOCAL passes retain Spark's existing shuffled offer order. + + 4.3.0 + spark.scheduler.revive.interval 1s