Skip to content

[SPARK-13279] Remove unnecessary duplicate check in addPendingTask fu… #11167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sitalkedia wants to merge 1 commit into from
Closed

[SPARK-13279] Remove unnecessary duplicate check in addPendingTask fu… #11167

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,9 +114,14 @@ private[spark] class TaskSetManager(
// treated as stacks, in which new tasks are added to the end of the
// ArrayBuffer and removed from the end. This makes it faster to detect
// tasks that repeatedly fail because whenever a task failed, it is put
// back at the head of the stack. They are also only cleaned up lazily;
// when a task is launched, it remains in all the pending lists except
// the one that it was launched from, but gets removed from them later.
// back at the head of the stack. These collections may contain duplicates
// for two reasons:
// (1): Tasks are only removed lazily; when a task is launched, it remains
// in all the pending lists except the one that it was launched from.
// (2): Tasks may be re-added to these lists multiple times as a result
// of failures.
// Duplicates are handled in dequeueTaskFromList, which ensures that a
// task hasn't already started running before launching it.
private val pendingTasksForExecutor = new HashMap[String, ArrayBuffer[Int]]

// Set of pending tasks for each host. Similar to pendingTasksForExecutor,
Expand Down Expand Up @@ -181,9 +186,7 @@ private[spark] class TaskSetManager(
private def addPendingTask(index: Int) {
// Utility method that adds `index` to a list only if it's not already there
def addTo(list: ArrayBuffer[Int]) {
if (!list.contains(index)) {
list += index
}
list += index
}
Copy link
Contributor

@markhamstra markhamstra Feb 12, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for not getting to this party until after the merge, but it strikes me that addTo is now only making a negative contribution toward understanding the code -- mostly because the "...if it's not already there" comment is now wrong. I don't see why the handful of uses of addTo shouldn't just be replaced with, e.g.:

pendingTasksForExecutor.getOrElseUpdate(e.executorId, new ArrayBuffer) += index

Copy link
Contributor

@kayousterhout kayousterhout Feb 12, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're in luck because I accidentally merged before Jenkins got to it, so we reverted it pending tests. There's a new PR here: #11175

Would you mind sticking this comment there? Agree with your sentiment.

Copy link
Author

@sitalkedia sitalkedia Feb 12, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated the PR : #11175 according to @markhamstra 's comment.


for (loc <- tasks(index).preferredLocations) {
Expand Down