Skip to content

Commit 1b1c849

Browse files
JoshRosenandrewor14
authored andcommitted
[SPARK-18640] Add synchronization to TaskScheduler.runningTasksByExecutors
## What changes were proposed in this pull request? The method `TaskSchedulerImpl.runningTasksByExecutors()` accesses the mutable `executorIdToRunningTaskIds` map without proper synchronization. In addition, as markhamstra pointed out in #15986, the signature's use of parentheses is a little odd given that this is a pure getter method. This patch fixes both issues. ## How was this patch tested? Covered by existing tests. Author: Josh Rosen <[email protected]> Closes #16073 from JoshRosen/runningTasksByExecutors-thread-safety. (cherry picked from commit c51c772) Signed-off-by: Andrew Or <[email protected]>
1 parent 8b33aa0 commit 1b1c849

File tree

3 files changed

+4
-4
lines changed

3 files changed

+4
-4
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ class SparkStatusTracker private[spark] (sc: SparkContext) {
112112
*/
113113
def getExecutorInfos: Array[SparkExecutorInfo] = {
114114
val executorIdToRunningTasks: Map[String, Int] =
115-
sc.taskScheduler.asInstanceOf[TaskSchedulerImpl].runningTasksByExecutors()
115+
sc.taskScheduler.asInstanceOf[TaskSchedulerImpl].runningTasksByExecutors
116116

117117
sc.getExecutorStorageStatus.map { status =>
118118
val bmId = status.blockManagerId

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ private[spark] class TaskSchedulerImpl(
9191
// IDs of the tasks running on each executor
9292
private val executorIdToRunningTaskIds = new HashMap[String, HashSet[Long]]
9393

94-
def runningTasksByExecutors(): Map[String, Int] = {
94+
def runningTasksByExecutors: Map[String, Int] = synchronized {
9595
executorIdToRunningTaskIds.toMap.mapValues(_.size)
9696
}
9797

core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -304,7 +304,7 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with L
304304
// Check that state associated with the lost task attempt is cleaned up:
305305
assert(taskScheduler.taskIdToExecutorId.isEmpty)
306306
assert(taskScheduler.taskIdToTaskSetManager.isEmpty)
307-
assert(taskScheduler.runningTasksByExecutors().get("executor0").isEmpty)
307+
assert(taskScheduler.runningTasksByExecutors.get("executor0").isEmpty)
308308
}
309309

310310
test("if a task finishes with TaskState.LOST its executor is marked as dead") {
@@ -335,7 +335,7 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with L
335335
// Check that state associated with the lost task attempt is cleaned up:
336336
assert(taskScheduler.taskIdToExecutorId.isEmpty)
337337
assert(taskScheduler.taskIdToTaskSetManager.isEmpty)
338-
assert(taskScheduler.runningTasksByExecutors().get("executor0").isEmpty)
338+
assert(taskScheduler.runningTasksByExecutors.get("executor0").isEmpty)
339339

340340
// Check that the executor has been marked as dead
341341
assert(!taskScheduler.isExecutorAlive("executor0"))

0 commit comments

Comments
 (0)