Skip to content

[SPARK-23040][CORE]: Returns interruptible iterator for shuffle reader #20449

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
advancedxy wants to merge 10 commits into from
+72 −2
Closed

[SPARK-23040][CORE]: Returns interruptible iterator for shuffle reader #20449

Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
acca0e3
[SPARK-23040][CORE]: Returns interruptible iterator for shuffle reader
advancedxy Jan 31, 2018
ddeffd8
Update comments and tests.
advancedxy Feb 9, 2018
88e86e0
Fix style issues and update comments
advancedxy Feb 26, 2018
ba2f355
Remove unnecessary Semaphore and Future block
advancedxy Feb 27, 2018
d6ed9a1
Use single partition in unit test
advancedxy Feb 27, 2018
8c15c56
Add more comments for the invocation of `f.cancel()`
advancedxy Feb 27, 2018
756e0b7
Hardcode numSlice(num of partition) to 1
advancedxy Mar 1, 2018
2061d0a
Reset global semaphores as they may interfere each other in multiple
advancedxy Mar 1, 2018
a3d8ad5
Use lowercase for test name.
advancedxy Mar 1, 2018
28119e9
Add comment for global variables
advancedxy Mar 1, 2018
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
Prev Previous commit
Next Next commit
Add more comments for the invocation of f.cancel()
  • Loading branch information
@advancedxy
advancedxy committed Feb 27, 2018
commit 8c15c564c7d2d0adc0cfd725e34dbd359c6a0ab6
6 changes: 6 additions & 0 deletions core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,12 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
})

taskStartedSemaphore.acquire()
Copy link
Contributor

@cloud-fan cloud-fan Feb 27, 2018

Choose a reason for hiding this comment

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

why not taskStartedSemaphore.acquire(numSlice)?

Copy link
Contributor Author

@advancedxy advancedxy Feb 27, 2018

Choose a reason for hiding this comment

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

As soon as one task starts, we can cancel the job.

// Job is cancelled when:
// 1. task in reduce stage has been started, guaranteed by previous line.
// 2. task in reduce stage is blocked after processing at most 10 records as
// taskCancelledSemaphore is not released until cancelTasks event is posted
// After job being cancelled, task in reduce stage will be cancelled and no more iteration are
// executed.
f.cancel()
Copy link
Contributor

@cloud-fan cloud-fan Feb 27, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

We should add some comment to explain when we reach here. From what I am seeing:

  1. taskStartedSemaphore.release() must be called, so at least one task is started.
  2. the first task has processed no more than 10 records, the second task hasn't processed any data, because the reduce stage is not finished and taskCancelledSemaphore.acquire() will be blocked.

Copy link
Contributor Author

@advancedxy advancedxy Feb 27, 2018

Choose a reason for hiding this comment

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

will do


val e = intercept[SparkException](f.get()).getCause
Expand Down