Sort shuffle read improvements

[jerryshao]

Hi @sryza , I've done some improvements in current SortSHuffleRead, would you mind help reviewing this and give me some comments, thanks a lot!

1. copy memory shuffle block from off-heap to on-heap if memory is allocated with direct memory to control the off-heap memory size.
2. Some code style modification and better failure processing logic.

[sryza]

Any particular reason for this change? Spark tends to prefer foreach to while loops outside of performance-critical regions.

[jerryshao]

Hi Sandy, there's a bug around here:

java.lang.IndexOutOfBoundsException: 1
        at scala.collection.mutable.ResizableArray$class.apply(ResizableArray.scala:43)
        at scala.collection.mutable.ArrayBuffer.apply(ArrayBuffer.scala:47)
        at scala.collection.mutable.ArrayBuffer.apply(ArrayBuffer.scala:47)
        at scala.collection.immutable.Range.foreach(Range.scala:141)
        at org.apache.spark.util.collection.TieredDiskMerger$DiskToDiskMerger.run(TieredDiskMerger.scala:188)

I'm guessing this Range.foreach may have some problems with queue.take. I changed to while loop to try again, seems the exception is gone.

[sryza]

Could the problem be that foreach is missing a parameter?

Does

(0 until mergeFactor).foreach { x=>
  blocksToMerge += onDiskBlocks.take()
}

work?

[jerryshao]

I didn't try this, I think this can be worked from my understanding. Meantime I think while is just the style choice, also very simple and straightforward here :)

[sryza]

Ok, not a big deal either way.

Do you think we're ready to submit this otherwise?

[sryza]

The second problem seems like it's a product of the first one, right? If we continually are unable to acquire memory, we are forced to spill.

[jerryshao]

From my understanding, this two problems may both related to the unexpected behavior of ShuffleMemoryManage. I have no small reproducible unit test to address this issue. Though I think it's not the problem of SortShuffleReader, it's worthy to investigate it.

[jerryshao]

Hi @sryza , seems this JIRA addresses the same problem as I mentioned before (https://issues.apache.org/jira/browse/SPARK-4452).

[sryza]

It still wouldn't explained why the shuffle memory manager reports negative memory though, right?

[jerryshao]

Yes, the first problem is actually a bug, while the second one seems like the same issue as SPARK-4452.

[sryza]

Had one small style comment. Otherwise this looks good to me.

[sryza]

Merging this