[GitHub] [spark] Ngone51 commented on a change in pull request #30650: [SPARK-24818][CORE] Support delay scheduling for barrier execution

[GitBox <gi...@apache.org>]

Ngone51 commented on a change in pull request #30650:
URL: https://github.com/apache/spark/pull/30650#discussion_r549324565



##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -439,77 +446,109 @@ private[spark] class TaskSetManager(
         }
       }
 
+      var dequeuedTaskIndex: Option[Int] = None
       val taskDescription =
         dequeueTask(execId, host, allowedLocality)
           .map { case (index, taskLocality, speculative) =>
-        // Found a task; do some bookkeeping and return a task description
-        val task = tasks(index)
-        val taskId = sched.newTaskId()
-        // Do various bookkeeping
-        copiesRunning(index) += 1
-        val attemptNum = taskAttempts(index).size
-        val info = new TaskInfo(taskId, index, attemptNum, curTime,
-          execId, host, taskLocality, speculative)
-        taskInfos(taskId) = info
-        taskAttempts(index) = info :: taskAttempts(index)
-        if (legacyLocalityWaitReset && maxLocality != TaskLocality.NO_PREF) {
-          resetDelayScheduleTimer(Some(taskLocality))
-        }
-        // Serialize and return the task
-        val serializedTask: ByteBuffer = try {
-          ser.serialize(task)
-        } catch {
-          // If the task cannot be serialized, then there's no point to re-attempt the task,
-          // as it will always fail. So just abort the whole task-set.
-          case NonFatal(e) =>
-            val msg = s"Failed to serialize task $taskId, not attempting to retry it."
-            logError(msg, e)
-            abort(s"$msg Exception during serialization: $e")
-            throw new TaskNotSerializableException(e)
-        }
-        if (serializedTask.limit() > TaskSetManager.TASK_SIZE_TO_WARN_KIB * 1024 &&
-          !emittedTaskSizeWarning) {
-          emittedTaskSizeWarning = true
-          logWarning(s"Stage ${task.stageId} contains a task of very large size " +
-            s"(${serializedTask.limit() / 1024} KiB). The maximum recommended task size is " +
-            s"${TaskSetManager.TASK_SIZE_TO_WARN_KIB} KiB.")
-        }
-        addRunningTask(taskId)
-
-        // We used to log the time it takes to serialize the task, but task size is already
-        // a good proxy to task serialization time.
-        // val timeTaken = clock.getTime() - startTime
-        val tName = taskName(taskId)
-        logInfo(s"Starting $tName ($host, executor ${info.executorId}, " +
-          s"partition ${task.partitionId}, $taskLocality, ${serializedTask.limit()} bytes) " +
-          s"taskResourceAssignments ${taskResourceAssignments}")
-
-        sched.dagScheduler.taskStarted(task, info)
-        new TaskDescription(
-          taskId,
-          attemptNum,
-          execId,
-          tName,
-          index,
-          task.partitionId,
-          addedFiles,
-          addedJars,
-          addedArchives,
-          task.localProperties,
-          taskResourceAssignments,
-          serializedTask)
-      }
+            dequeuedTaskIndex = Some(index)
+            if (legacyLocalityWaitReset && maxLocality != TaskLocality.NO_PREF) {
+              resetDelayScheduleTimer(Some(taskLocality))

Review comment:
       This's the legacy delay scheduling behavior(disabled by default). In the new delay scheduling, we do reset only if all barrier tasks get scheduled:
   https://github.com/apache/spark/blob/e3058ba17cb4512537953eb4ded884e24ee93ba2/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala#L589-L593
   
   (For the barrier taskset, `noDelaySchedulingRejects` means all tasks get scheduled in a single resourceOffers round.)
   
   
   For the legacy delay scheduling, I'd prefer to keep the same behavior as it is.
   




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org