Skip to content

[SPARK-21527][CORE] Use buffer limit in order to use JAVA NIO Util's buffercache #18730

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
caneGuy wants to merge 19 commits into from
Closed

[SPARK-21527][CORE] Use buffer limit in order to use JAVA NIO Util's buffercache #18730

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
7cbadc5
[SPARK][CORE] Slice write by channel
caneMi Jul 25, 2017
fc91f96
Fix code style
caneGuy Aug 23, 2017
bab91db
Refine to avoid underflow
caneGuy Aug 24, 2017
72aef67
Fix no ending loop
caneGuy Aug 24, 2017
4789772
Remove useless unit test
caneGuy Aug 24, 2017
aeabe1d
Delete unused package
caneGuy Aug 24, 2017
ca77b51
Add benchmark testing with Utils.BenchMark
caneGuy Aug 24, 2017
e84a6d7
Refine code style
caneGuy Aug 24, 2017
d708142
Update as comment replace writeFully with slice logic
caneGuy Aug 24, 2017
33a2796
Remove unused package
caneGuy Aug 24, 2017
ab384d4
No need call toInt
caneGuy Aug 24, 2017
b669351
Only need to use bytes.remaining()
caneGuy Aug 24, 2017
717f886
Use config for NIO_BUFFER_LIMIT
caneGuy Aug 24, 2017
9d3004d
Use BUFFER_WRITE_CHUNK_SIZE
caneGuy Aug 24, 2017
e48f44f
Update as commented
caneGuy Aug 24, 2017
fc184aa
Fix build error
caneGuy Aug 24, 2017
f1d67a3
Fix compile error and refine scala style
caneGuy Aug 24, 2017
14ca824
Need to convert to int to fix compile error
caneGuy Aug 25, 2017
a02575e
Avoid per-loop type cast
caneGuy Aug 25, 2017
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
Avoid per-loop type cast
  • Loading branch information
@caneGuy
caneGuy committed Aug 25, 2017
commit a02575e3bb3f5f7f2f2d9f7a0a51a1f162c12c4d
2 changes: 2 additions & 0 deletions core/src/main/scala/org/apache/spark/internal/config/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,6 +298,8 @@ package object config {
.internal()
.doc("The chunk size during writing out the bytes of ChunkedByteBuffer.")
.bytesConf(ByteUnit.BYTE)
.checkValue(_ <= Int.MaxValue, "The chunk size during writing out the bytes of" +
" ChunkedByteBuffer should not larger than Int.MaxValue.")
.createWithDefault(64 * 1024 * 1024)
Copy link
Contributor

@cloud-fan cloud-fan Aug 25, 2017

Choose a reason for hiding this comment

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

add a checkValue to make sure the value is smaller than Int.Max


private[spark] val CHECKPOINT_COMPRESS =
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
// Chunk size in bytes
private val bufferWriteChunkSize =
Option(SparkEnv.get).map(_.conf.get(config.BUFFER_WRITE_CHUNK_SIZE))
.getOrElse(config.BUFFER_WRITE_CHUNK_SIZE.defaultValue.get)
.getOrElse(config.BUFFER_WRITE_CHUNK_SIZE.defaultValue.get).toInt

private[this] var disposed: Boolean = false

Expand All @@ -65,7 +65,7 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
for (bytes <- getChunks()) {
while (bytes.remaining() > 0) {
val ioSize = Math.min(bytes.remaining(), bufferWriteChunkSize)
bytes.limit(bytes.position + ioSize.toInt)
bytes.limit(bytes.position + ioSize)
channel.write(bytes)
}
}
Expand Down