LuciferYang commented on code in PR #57274:
URL: https://github.com/apache/spark/pull/57274#discussion_r3600630446
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2996,6 +2996,41 @@ class TaskSetManagerSuite
s"Expected staleMapIndexes to contain mapIndex 0, got $staleMapIndexes")
}
+ test("SPARK-58137: hasAttemptOnHost uses taskAttemptHosts for O(1) host
lookup") {
+ sc = new SparkContext("local", "test")
+ sc.conf.set(config.SPECULATION_MULTIPLIER, 0.0)
+ sc.conf.set(config.SPECULATION_ENABLED, true)
+ sc.conf.set(config.SPECULATION_QUANTILE, 0.5)
+ sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"),
+ ("exec3", "host3"))
+ val taskSet = FakeTask.createTaskSet(1, Seq(TaskLocation("host1",
"exec1")))
+ val clock = new ManualClock()
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES, clock
= clock)
+
+ // Launch the single task on host1 (exec1)
+ val task1 = manager.resourceOffer("exec1", "host1",
TaskLocality.PROCESS_LOCAL)._1.get
+ assert(task1.index === 0)
+ assert(manager.runningTasks === 1)
+
+ // Mark the task as speculatable and add to pending speculative list
+ manager.speculatableTasks += 0
+ manager.addPendingTask(0, speculatable = true)
+
+ // Speculative task should NOT be scheduled on host1 which already has an
attempt
+ assert(manager.resourceOffer("exec1", "host1",
TaskLocality.ANY)._1.isEmpty)
+
+ // Speculative task SHOULD be scheduled on host2 which has no attempt for
task 0
+ val task2 = manager.resourceOffer("exec2", "host2", TaskLocality.ANY)._1
+ assert(task2.isDefined)
+ assert(task2.get.index === 0)
+ assert(task2.get.attemptNumber === 1)
+
+ // Now task 0 has attempts on both host1 and host2.
+ // Offering host1 or host2 should not schedule another speculative copy.
+ assert(manager.resourceOffer("exec1", "host1",
TaskLocality.ANY)._1.isEmpty)
Review Comment:
The last two asserts (offer host1/host2 → empty) don't actually exercise
`hasAttemptOnHost`. After the host2 offer schedules attempt 1, the speculative
copy is dequeued so `speculatableTasks` is empty and `copiesRunning(0) == 2`;
from there the speculative branch short-circuits regardless of host, so these
asserts pass even if `hasAttemptOnHost` always returns `false`. The
discriminating asserts are the earlier two: host1 → empty (an attempted host is
skipped) and host2 → scheduled (a fresh host is allowed). The comment "Offering
host1 or host2 should not schedule another speculative copy" attributes the
emptiness to the host check, but it's really the exhausted speculatable list.
To make the last asserts meaningful, bring the task back to `copiesRunning ==
1` (e.g. fail one attempt) so the host check is the only gate; otherwise they
can be dropped.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -545,6 +547,10 @@ private[spark] class TaskSetManager(
taskInfos(taskId) = info
executorIdToTaskIds.getOrElseUpdate(execId, new
OpenHashSet[Long]).add(taskId)
taskAttempts(index) = info :: taskAttempts(index)
+ if (taskAttemptHosts(index) == null) {
Review Comment:
`taskAttemptHosts` is written on every launch in `prepareLaunchingTask`, but
`hasAttemptOnHost` (its only reader) is reachable only on the speculative
dequeue path, and that path is dead unless `spark.speculation` is enabled,
which it is not by default: `speculatableTasks` stays empty,
`dequeueTaskHelper` early-returns at `:390`, and `hasAttemptOnHost` is never
called. So in the default config every launch writes a host into (and on the
first attempt of each index allocates) a structure nothing reads.
`speculationEnabled` is already a field on this class (`:80`); wrapping just
the populate block (`:550-553`) in `if (speculationEnabled)`, while leaving the
array at `:134` allocated as-is, preserves behavior — `hasAttemptOnHost` then
reads default-null elements (returns `false`, exactly as today) and the read
path needs no change. One caveat: don't null the array itself. The existing
`hosts != null` guard is element-level, so a null array would NPE at
`taskAttemptHosts(taskIndex)`
(`:358`) before that check; to drop the array too you'd need an extra
`taskAttemptHosts != null` guard at the top of `hasAttemptOnHost`. The
per-launch cost is small in absolute terms, but it adds cost to the always-on
default path to speed up a path that is off by default, and the fix is nearly
free.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2996,6 +2996,41 @@ class TaskSetManagerSuite
s"Expected staleMapIndexes to contain mapIndex 0, got $staleMapIndexes")
}
+ test("SPARK-58137: hasAttemptOnHost uses taskAttemptHosts for O(1) host
lookup") {
Review Comment:
The test name references the implementation and complexity ("uses
taskAttemptHosts for O(1) host lookup"), but the test can only observe
scheduling behavior, not the field used or the big-O. Something like
"speculative task is not scheduled on a host that already ran an attempt"
matches what's asserted and won't go stale if the internals change.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -131,6 +131,7 @@ private[spark] class TaskSetManager(
private val killedByOtherAttempt = new HashSet[Long]
val taskAttempts = Array.fill[List[TaskInfo]](numTasks)(Nil)
+ val taskAttemptHosts = Array.ofDim[HashSet[String]](numTasks)
Review Comment:
`taskAttemptHosts` is a public `val`, but nothing outside `TaskSetManager`
reads it (the sibling `taskAttempts` is public only because tests read it). The
in-file precedent for a purely-internal per-index array is `private val
numFailures`. I'd make it `private[scheduler]` (or `private`) to keep it off
the class's public API.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
