Skip to content

[SPARK-2670] FetchFailedException should be thrown when local fetch has failed #1578

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
sarutak wants to merge 10 commits into from
Closed

[SPARK-2670] FetchFailedException should be thrown when local fetch has failed #1578

Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e310c0b
Modified BlockFetcherIterator to handle local fetch failure as fatch …
sarutak Jul 24, 2014
460dc01
Merge branch 'master' of git://git.apache.org/spark into SPARK-2670
sarutak Jul 29, 2014
a3a9be1
Modified BlockFetcherIterator for SPARK-2670
sarutak Jul 29, 2014
b7b8250
Modified BasicBlockFetchIterator to fail fast when local fetch error …
sarutak Jul 29, 2014
4fca130
Added test cases for BasicBlockFetcherIterator
sarutak Jul 29, 2014
5d05855
Merge branch 'master' of git://git.apache.org/spark into SPARK-2670
sarutak Jul 31, 2014
03bcb02
Merge branch 'SPARK-2670' of github.com:sarutak/spark into SPARK-2670
sarutak Jul 31, 2014
d353984
Refined assertion messages in BlockFetcherIteratorSuite.scala
sarutak Jul 31, 2014
e8713cc
Merge branch 'master' of git://git.apache.org/spark into SPARK-2670
sarutak Jul 31, 2014
85c8938
Removed useless results.put for fail fast
sarutak Jul 31, 2014
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
21 changes: 14 additions & 7 deletions core/src/main/scala/org/apache/spark/storage/BlockFetcherIterator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,14 +200,21 @@ object BlockFetcherIterator {
// these all at once because they will just memory-map some files, so they won't consume
// any memory that might exceed our maxBytesInFlight
for (id <- localBlocksToFetch) {
getLocalFromDisk(id, serializer) match {
case Some(iter) => {
// Pass 0 as size since it's not in flight
results.put(new FetchResult(id, 0, () => iter))
logDebug("Got local block " + id)
try{
Copy link
Contributor

@mateiz mateiz Jul 25, 2014

Choose a reason for hiding this comment

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

Small code style thing, add a space before the {

getLocalFromDisk(id, serializer) match {
case Some(iter) => {
// Pass 0 as size since it's not in flight
results.put(new FetchResult(id, 0, () => iter))
logDebug("Got local block " + id)
}
case None => {
throw new BlockException(id, "Could not get block " + id + " from local machine")
}
}
case None => {
throw new BlockException(id, "Could not get block " + id + " from local machine")
} catch {
case e: Exception => {
logError(s"Error occurred while fetch local block $id", e)
results.put(new FetchResult(id, -1, null))
Copy link
Contributor

@mateiz mateiz Jul 30, 2014

Choose a reason for hiding this comment

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

I wouldn't do drop and such on a ConcurrentQueue, since it might drop stuff other threads were adding. Just do a results.put on the failed block and don't worry about dropping other ones. You can actually move the try/catch into the for loop and add a "return" at the bottom of the catch after adding this failing FetchResult.

Copy link
Member Author

@sarutak sarutak Jul 31, 2014

Choose a reason for hiding this comment

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

Thank you for your comment, @mateiz .

I wouldn't do drop and such on a ConcurrentQueue, since it might drop stuff other threads were adding. Just do a results.put on the failed block and don't worry about dropping other ones. You can actually move the try/catch into the for loop and add a "return" at the bottom of the catch after adding this failing FetchResult.

But, if it returns from getLocalBlocks immediately rest of FetchResults is not set to results, and we waits on results.take() in next method forever right? results is a instance of LinkedBlockingQueue and take method is blocking method.

Copy link
Contributor

@mateiz mateiz Jul 31, 2014

Choose a reason for hiding this comment

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

I thought next() would return a failure block, and then the caller of BlockFetcherIterator will just stop. Did you see it not doing that? I think all you have to do is put one FetchResult with size = -1 in the queue and return, and everything will be fine.

Copy link
Member Author

@sarutak sarutak Jul 31, 2014

Choose a reason for hiding this comment

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

I thought wrong. Exactly, in current usage of BlockFetcherIterator, next() is not invoked after FetchFailedException has been thrown.
I wonder it's a little bit problem that we can invoke next() after FetchFailedException even if there are no such usages in current implementation.
I think it's better to prohibit invoking next() after FetchFailedException to clearly express the correct usage of the method.

}
Copy link
Contributor

@mateiz mateiz Jul 25, 2014

Choose a reason for hiding this comment

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

Why do we throw an exception above and then immediately catch it, instead of doing results.put above? Is there any other kind of error that can happen beyond getLocalFromDisk returning None?

Also, the current code seems to forget the exception: it just puts in a failed result. Is this intentional, i.e. will get a FetchFailedException later? It seems we should return from this method ASAP if there's a problem.

Copy link
Member Author

@sarutak sarutak Jul 26, 2014

Choose a reason for hiding this comment

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

Actually, getLocalFromDisk never return None but can throw BlockException. so I think "case None" block above is useless and we should remove the "case None" block rather than doing results.put.

Is there any other kind of error that can happen beyond getLocalFromDisk returning None?

Yes, BlockException is thrown from getLocalFromDisk, and FileNotFoundException from DiskStore#getBytes when it failed to fetch shuffle____ from local disk.

Also, the current code seems to forget the exception: it just puts in a failed result. Is this intentional, i.e. will get a FetchFailedException later?

It's for get FetchFailedException later. If we return from BasicBlockFetchIterator#getLocallocks, we can't know whether rest of blocks can be read successfully or not.

}
}
Expand Down