mridulm commented on code in PR #36162:
URL: https://github.com/apache/spark/pull/36162#discussion_r886264430
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala:
##########
@@ -853,8 +857,11 @@ private[spark] class TaskSchedulerImpl(
// (taskId, stageId, stageAttemptId, accumUpdates)
val accumUpdatesWithTaskIds: Array[(Long, Int, Int, Seq[AccumulableInfo])]
= {
accumUpdates.flatMap { case (id, updates) =>
- val accInfos = updates.map(acc => acc.toInfo(Some(acc.value), None))
Option(taskIdToTaskSetManager.get(id)).map { taskSetMgr =>
+ val (accInfos, taskProgressRate) =
getTaskAccumulableInfosAndProgressRate(updates)
Review Comment:
Having said that, I was thinking along lines of the following:
1. For successful tasks, as part of TSM.`handleSuccessfulTask`, update the
metrics required for efficiency computation.
1. That is, in `handleSuccessfulTask`, we maintain state required for
efficiency computation - without needing to recompute it for all tasks in
`maybeRecompute` -> just do it for running tasks.
2. This is similar to what we do with medianHeap.
2. With this in place, I was not sure if we want to maintain the state in
`TaskInfo` anymore.
1. We could simply maintain a concurrent map for running tasks ->
efficiency details in `InefficientTaskCalculator`
2. In `executorHeartbeatReceived`, we could simply delegate to TSM to
update its calculator with the state given task id.
Thoughts ?
--
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: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
